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


def init_send(s_port:serial.Serial, send_str):
    s_info = bytearray()
    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
        # m_ram = re.search("Recieving RAM-IMAGE in xmodem : C", s_info.decode("utf-8"))
        if(s_info.find(b"Recieving RAM-IMAGE in xmodem : C")>=0):
            m_ram=True
        else:
            m_ram=False

        if m_ram:
            print ("Program enters transmission mode...")
            break


def burn_ram_bin(x_modem:xmodem.XMODEM, r_file):
    global trans_time_0
    wf = open(log_file, 'a')
    stime = datetime.datetime.now()
    print ("Transferring %s..." % r_file)
    wf.writelines("Transferring %s..." % r_file)
    try:
        stream = open(r_file, 'rb')
    except Exception:
        print("Cannot load file, please check the file path and retry. Press <enter> to exit")
        sys.exit()

    xmodem_send = x_modem.send(stream, callback=status_update_0)
    etime = datetime.datetime.now()
    trans_time_0 = (etime - stime).seconds
    print ("\nTransferring ram.bin result: %s, consuming time: %s s \n" % (xmodem_send, trans_time_0))
    wf.writelines("Transferring %s..." % r_file)
    wf.close()

def burn_flash_bin(s_port:serial.Serial, x_modem:xmodem.XMODEM, f_file):
    global trans_time_1
    s_info = bytearray()
    wf = open(log_file, 'a')
    while True:
        s_info += s_port.read(1)
        print(f"recv:{s_info.hex(' ')}")
        bytes2read = s_port.in_waiting
        tmp = s_port.read(bytes2read)
        s_info += tmp
        if(s_info.find(b'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:
            # print m_flash.group(0)
            print (r"Recieving FLASH-IMAGE in xmodem : C")
            wf.writelines("Recieving FLASH-IMAGE in xmodem : C")
            s_info = bytearray()
            stime = datetime.datetime.now()
            print ("Transferring %s..." % f_file)
            wf.writelines("Transferring %s..." % f_file)
            try:
                stream = open(f_file, 'rb')
            except Exception:
                print("Cannot load file, please check the file path and retry. Press <enter> to exit")
                sys.exit()

            xmodem_send = x_modem.send(stream, quiet=True, callback=status_update_1)
            etime = datetime.datetime.now()
            trans_time_1 = (etime - stime).seconds
            print ("\nTransferring iot_flash.bin result: %s, consuming time: %d s \n" % (xmodem_send, trans_time_1))
            wf.writelines("\nTransferring iot_flash.bin result: %s, consuming time: %d s \n" % (xmodem_send, trans_time_1))
        elif m_done:
            print (m_done.group(0))
            break
        else:
            pass
    wf.close()


def upload_callback(total_packets, success_count, error_count, packet_size):
    # callback(total_packets, success_count, error_count, packet_size)
    print("upload:",total_packets, success_count, error_count, packet_size)

# 上传固件
def upload_bin(x_modem:xmodem.XMODEM, w_file):
    global trans_time_0
    wf = open(log_file, 'a')
    stime = datetime.datetime.now()
    print ("Transferring %s..." % w_file)
    wf.writelines("Transferring %s..." % w_file)
    try:
        stream = open(w_file, 'wb+')
    except Exception:
        print("Cannot load file, please check the file path and retry. Press <enter> to exit")
        sys.exit()

    xmodem_send = x_modem.recv(stream, callback=upload_callback)
    etime = datetime.datetime.now()
    trans_time_0 = (etime - stime).seconds
    print ("\nTransferring ram.bin result: %s, consuming time: %s s \n" % (xmodem_send, trans_time_0))
    wf.writelines("Transferring %s..." % w_file)
    wf.close()




def getc(size, timeout=1):
    data=ser.read(size)
    print("getc:",data.hex(' '))
    return data or None


def putc(data, timeout=1):
    # print("putc:",data.hex(' '))
    ser.write(data)
    time.sleep(0.03)


def status_update_0(total_packets, success_count, error_count):
    print ("total_packets: %s, success_count: %s, error_count: %d" % (total_packets, success_count, error_count))
    # if total_packets % 10 == 0:
    #     print ('.'),


def status_update_1(total_packets, success_count, error_count):
    print ("total_packets: %s, success_count: %s, error_count: %d" % (total_packets, success_count, error_count))
    # if total_packets % 50 == 0:
    #     print ('.'),

log_file = "log.txt"
if __name__ == '__main__':

    initial_info = """
    *** Version 11.0.0.1
    *** This is the tool for chip files transmission
    *** You can modify xmodem_config.txt to configure
    *** Serial Port, Baud Rate, Transfer Files Path
    *** Press RST Button to continue the operation...
    """
    print (initial_info)
    wf = open(log_file, 'w+')
    wf.writelines(initial_info)
    exit_flag = 0
    trans_time_0, trans_time_1 = 0, 0
    config_file = r"xmodem_config.txt"
    init_str, sp_num, b_rate, nb_rate, ram_file, iot_flash_file, ser = None, None, None, None, None, None, None
    for eachline in fileinput.FileInput(config_file):
        m_info = re.match(r"(\w+)\s*=\s*(.+)", eachline)
        if m_info:
            if m_info.group(1) == "init_str":
                init_str = m_info.group(2)
            elif m_info.group(1) == "serial_port_num":
                sp_num = int(m_info.group(2))
            elif m_info.group(1) == "baud_rate":
                b_rate = int(m_info.group(2))
            elif m_info.group(1) == "new_baud_rate":
                nb_rate = int(m_info.group(2))
            elif m_info.group(1) == "ram_file":
                ram_file = m_info.group(2)
            elif m_info.group(1) == "iot_flash_file":
                iot_flash_file = m_info.group(2)
            else:
                pass

    try:
        ser = serial.Serial(port='COM' + str(sp_num), baudrate=b_rate, timeout=0.3)
    except Exception:
        print("Serial Port COM%s Conflicts!!! Press <enter> to Close it and retry..." % str(sp_num))
        wf.writelines("Serial Port COM%s Conflicts!!! Press <enter> to Close it and retry..." % str(sp_num))
        sys.exit()
    modem = xmodem.XMODEM(getc, putc, mode='xmodem1k')

    # 发送启动字符让设备进入xmodem模式
    # init_send(ser, init_str.encode("utf-8"))
    # burn_ram_bin(modem, ram_file)

    # ser.baudrate=nb_rate
    time.sleep(0.5)

    # burn_flash_bin(ser, modem, iot_flash_file)
    # 显示flash信息
    ser.write(b"f i\n")
    time.sleep(0.5)
    print(ser.read(4096).decode('utf-8'))
    # 显示image信息
    ser.write(b"f s\n")
    time.sleep(0.5)
    print(ser.read(4096).decode('utf-8'))
    # 上传整个镜像
    ser.write(b"fw u d all\n")
    time.sleep(0.5)
    print(ser.read(4096).decode('utf-8'))
    # upload_bin(modem,"upload.bin")

    print ("Total transmission time: %s s" % str(trans_time_0+trans_time_1))
    wf.writelines("Total transmission time: %s s" % str(trans_time_0+trans_time_1))
    wf.close()