xref: /developtools/integration_verification/DeployDevice/src/func/liteOsUpgrade/liteOsUpgrade_L2_app.py

#-*- coding:utf-8 -*-
import uuid
import sys
import subprocess
import os
import serial
import datetime

from core.base import BaseApp, dec_stepmsg
from util.file_locker import FileLock
from util.log_info import logger
from util.time_info import get_now_time_str_info, get_now_time_info, Timeout, timeout
from aw.Telnet.TelnetClient import TelConnect
from aw.Common.Constant import CONSTANT
from aw.Download.Download import *
from aw.Common.Common import getHostIp, copyFile, copyDirectory
from aw.ExtractFile.ExtractFile import *
from aw.poweronoff.serial_power_on_off import usbPowerOnOff, usbPowerOnOffV2
from threading import Thread
from subprocess import getstatusoutput

lock_suffix = CONSTANT.File.LOCK_SUFFIX  #通过文件锁实现并发下载
suc_file = CONSTANT.File.SUC_FILE        #通过本文件是区分版本是否成功下载
failed_file = CONSTANT.File.FAILED_FILE  #通过本文件是标记文件下载失败
READ_MAXTIMEOUT = 5
READ_TIMEOUT = 5
READ_MINITIMEOUT = 2
uboot_finish = 'hisilicon #'
cmd_finish = ' #'
error_str_list = ['Unknown', '发送起始帧失败', '发送头帧失败']

_is_download_success = False


class liteOsUpgrade_L2(BaseApp):

    def __init__(self, param_file):
        super().__init__(param_file)
        self.param_List = ["deploy_com",
                        "usb_port",
                           "upgrade_upgradeLocation"]

    @dec_stepmsg("hongmeng L2 flash")
    def excute(self):
        '''
        #===================================================================================
        #   @Method:        excute(self)
        #   @Precondition:  none
        #   @Func:          升级执行入口
        #   @PostStatus:    none
        #   @eg:            excute()
        #   @return:        True or Flase
        #===================================================================================
        '''
        step_index = self.params_dict.get("step_list").index("liteOsUpgrade_L2_app")

        # 执行下载
        try:
            if not self.download():
                CONSTANT.ENVERRMESSAGE = "image download fail"
                logger.printLog(CONSTANT.ENVERRMESSAGE)
                return False
        except Exception as e:
            raise e


        # 执行升级
        try:
            if not self.upgrade():
                CONSTANT.ENVERRMESSAGE = "board upgrade fail"
                logger.printLog(CONSTANT.ENVERRMESSAGE)
                return False
            return True
        except Exception as e:
            raise e

    @dec_stepmsg("download")
    @timeout(18000)
    def download(self):
        '''
        #===================================================================================
        #   @Method:        download(self)
        #   @Precondition:  none
        #   @Func:          构建下载到本地的路径，执行相应包的下载
        #   @PostStatus:    none
        #   @eg:            download()
        #   @return:        True or Flase
        #===================================================================================
        '''
        global version_savepath, version_name, _is_download_success
        dir_path = CONSTANT.Path.getDirPath()
        if self.params_dict.get("pbiid"):
            version_path = self.params_dict.get("pbiid")
            version_name = str(uuid.uuid5(uuid.NAMESPACE_URL, str(self.params_dict.get("pbiid")) + "FASTBOOT"))
            version_savepath = os.path.join(dir_path, self.params_dict.get("flash_type"), version_name)
        else:
            version_path = self.params_dict.get("upgrade_upgradeLocation")
            version_name = str(uuid.uuid5(uuid.NAMESPACE_URL, (self.params_dict.get("upgrade_upgradeLocation"))))
            version_savepath = os.path.join(dir_path, version_name)

        if self.params_dict.get("isDownload") == "True":
            logger.printLog("不需要做下载，直接返回")
            return True

        #执行img下载
        import hashlib
        save_file_str = version_path.replace("/", "").replace("\\", "")
        save_file_name = hashlib.sha1(save_file_str.encode("utf-8")).hexdigest()
        logger.info("download hash value:%s" % (save_file_name))
        save_path_file = os.path.join(dir_path, "record", "%s%s" % (save_file_name, ".txt"))
        logger.printLog('save_path_file: %s' % save_path_file)
        if not self.excutedown(version_path, os.path.join(version_savepath, "img"), save_path_file, False):
            logger.error("download img fail")
            return False
        _is_download_success = True
        #保存本地版本路径给devicetest去版本路径下取用例
        saveVersion(save_path_file, os.path.join(version_savepath, "img"))

        return True

    def excutedown(self, source_path, download_dir, suc_mark, is_file):
        '''
        #===================================================================================
        #   @Method:        excutedown(source_path, download_dir, is_file)
        #   @Precondition:  none
        #   @Func:          执行下载动作
        #   @PostStatus:    none
        #   @Param:         source_path：资源文件路径
        #                   download_dir:文件下载到本地的文件夹路径
        #                   is_file:是否是文件
        #
        #   @eg:            excutedown("xxxx", "D:\\local\\image", Flase， os_method)
        #   @return:        True or Flase
        #===================================================================================
        '''
        failed_mark = os.path.join(download_dir, failed_file)
        lock_path = os.path.join(download_dir, lock_suffix)
        file_lock = FileLock()

        if isDownLoadSuccess(download_dir, suc_mark, failed_mark):
            return True
        try:
            nowtime = get_now_time_str_info()
            logger.printLog("%s Downloading, please wait" % nowtime)
            file_lock.lockFile(lock_path)
            ret = ""
            logger.info("Get lock. Start to ")
            if self.params_dict.get("bt_enable") and self.params_dict.get("bt_enable") == "True":
                ret = downloadByBitComet(source_path, download_dir, os_method)
            elif source_path.startswith('\\\\'):
                ret = downloadByCopy(source_path, download_dir, is_file)
            elif self.params_dict.get("pbiid"):
                ret = downlaodByDownloadTool(version_savepath, self.params_dict.get("version_type"), "FASTBOOT", self.params_dict.get("pbiid"))
            elif source_path.startswith("http"):
                ret = downloadFileFromDevCloud(source_path, "", "", download_dir)

            if source_path.endswith(".zip"):
                zip_name = os.path.basename(source_path)
                ret = extractZipFile(os.path.join(download_dir, zip_name), download_dir)
            if source_path.endswith(".tar.gz") or (source_path.startswith("http") and ("file_id=" in source_path)):
                if source_path.startswith("http") and ("file_id=" in source_path):
                    if source_path.endswith(".tar.gz"):
                        zip_name = source_path.split('=')[-1]
                    else:
                        zip_name = "out.tar.gz"
                else:
                    zip_name = os.path.basename(source_path)
                logger.printLog(f'tar_file:{os.path.join(download_dir, zip_name)},dest_file:{download_dir}')
                ret = unTarFile(os.path.join(download_dir, zip_name), download_dir)
            nowtime = get_now_time_str_info()
            logger.printLog("%s download to %s end" % (nowtime, download_dir))

            if not ret:
                with open(failed_mark, "a+") as fp:
                    fp.write("")
            return ret
        except Exception as e:
            logger.printLog(e)
            raise Exception(e)
        finally:
            file_lock.releaseFile()


    @dec_stepmsg("upgrade")
    #@timeout(900)
    def upgrade(self):
        '''
        #===================================================================================
        #   @Method:        upgrade(self)
        #   @Precondition:  none
        #   @Func:          升级相关业务逻辑
        #   @PostStatus:    none
        #   @eg:            upgrade()
        #   @return:        True or Flase
        #===================================================================================
        '''
        global _is_download_success
        logger.printLog('开始升级')
        deploy_com = self.params_dict.get("deploy_com")
        usb_port = self.params_dict.get("usb_port")
        baudrate = self.params_dict.get("baudrate")
        #芯片类型,根据芯片类型获取对应的刷机命令
        flash_type = self.params_dict.get("flash_type")
        burn_usbport = self.params_dict.get("hiburn_usbport")
        device_ip = self.params_dict.get("Device_IP")
        device_netmask = self.params_dict.get("Device_Netmask")
        device_gatewayip = self.params_dict.get("Device_GatewayIP")
        chip_version = self.params_dict.get('chip_version')
        chip_version = chip_version.lower() if chip_version else chip_version
        sn = self.params_dict.get('sn')

        if not deploy_com:
            logger.error("deploy_com is NULL !!")
            return False
        if not burn_usbport:
            logger.error("hiburn_usbport is NULL !!")
            return False
        if not baudrate:
            baudrate = 115200
        #执行下载
        t_download = Thread(target=self.download, args=())
        t_download.start()
        scriptpath = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(os.path.dirname(__file__)))))
        #升级需要的工具归档
        toolworkspace = CONSTANT.OSType.getworkspace()
        hiburntoolpath = os.path.join(toolworkspace, "HiBurnCmdLine", "usb%s_tool" % burn_usbport)
        logger.info("hiburn tool path is: %s" % hiburntoolpath)
        if not os.path.exists(hiburntoolpath):
            if not burn_usbport:
                logger.error("hiburn_usbport is NULL !!")
                return False
            os.makedirs(hiburntoolpath)
            toolpath = os.path.join(scriptpath, "resource", "HiBurnCmdLine.zip")
            logger.info("copy %s to %s" % (toolpath, hiburntoolpath))
            copyFile(toolpath, hiburntoolpath)
            zip_name = os.path.basename(toolpath)
            ret = extractZipFile(os.path.join(hiburntoolpath, zip_name), hiburntoolpath)
            if ret:
                logger.info("unzip to %s succ" % (hiburntoolpath))
                #修改burn.config中的usb口
                configpath = os.path.join(hiburntoolpath, "config", "burn.config")
                all_data = ""
                with open(configpath, "r", encoding="utf-8") as cf:
                    for line in cf:
                        if "usbDeviceNumber=" in line:
                            old_str = line
                            line = line.replace(old_str, "usbDeviceNumber=%s\r\n" % burn_usbport)
                            logger.info("replace line: %s " % line)
                        all_data += line
                with open(configpath, "w", encoding="utf-8") as wf:
                    wf.write(all_data)
            else:
                logger.error("%s is not exit" % hiburntoolpath)
                return False
        os.chdir(hiburntoolpath)
        if flash_type.lower() == "ev300":
            chip_type = "Hi3518EV300"
        elif flash_type.lower() == "dv300":
            chip_type = "Hi3516DV300"
        else:
            logger.error("flash_type is : %s " % flash_type)
            return False
        #擦除fastboot
        logger.printLog('erase fastboot')
        flash_uboot_xml = os.path.join(scriptpath, "resource", "L2", flash_type.lower(), "flash_fastboot.xml")
        cmd = ".\jre\\bin\java -jar hiburn.jar --erase -n %s -m serial %s -x %s" % (chip_type, deploy_com.upper(), flash_uboot_xml)
        self.eraseDevice(cmd, usb_port)
        t_download.join()
        # 检查下载是否成功
        if not _is_download_success:
            return False

        #将升级需要的文件拷贝到镜像里面
        local_image_path = os.path.join(version_savepath, "img")
        old_xml_path = os.path.join(scriptpath, "resource", "L2", flash_type, "Hi3516DV300-emmc.xml")
        xml_path = os.path.join(local_image_path, "Hi3516DV300-emmc.xml")
        copyFile(old_xml_path, xml_path)
        # copyFile(ubootpath, local_image_path)
        scriptfile = os.path.join(scriptpath, "resource", "L2", f'{flash_type.lower()}', "update.txt")
        logger.info(f'scriptfile:{scriptfile}')

        logger.printLog('进行烧写')
        retry = 0
        while retry < 3:
            #usb刷机
            cmd = ".\jre\\bin\java -jar hiburn.jar --burn -n %s -m USBBootrom -x %s" % (chip_type, xml_path)
            logger.info("cmd is: %s" % cmd)
            ret, outpri = subprocess.getstatusoutput(cmd)
            logger.info("usb upgrade result: %s " % ret)
            logger.info("print console: %s " % outpri)
            if ret != 0:
                if ret == 4 and retry < 2:
                    time.sleep(10)
                    retry = retry + 1
                    logger.info('flash fail,so flash once again')
                    continue
                logger.info(ret)
                logger.error("hiburn usb upgrade failed!!")
                return False
            retry = retry + 3
        os.chdir(current_path)
        logger.info("hiburn upgrade end, check board status")
        logger.info("为保持hdc稳定性，等待60秒")
        time.sleep(60)
        try:
            #pass
            self.hdc_set_time(sn)
        except Exception as e:
            logger.printLog('hdc设置时间超时')
            logger.printLog(str(e))
        try:
            logger.info("打开serial")
            ser = serial.Serial(deploy_com, int(baudrate), timeout=0)
            if ser.is_open == False:
                ser.open()
            logger.info("get device status")
            board_type = getBoardType(ser)
            if board_type == "OHOS" and self.checkStatus(ser):
                return True

                # ret = sendCmd(ser, 'sa;reset;', READ_MAXTIMEOUT)
                # board_type = getBoardType(ser)
                # if board_type != "OHOS":
                #     logger.error("upgrade fail")
                #     return False
            else:
                return False
        except Exception as e:
            logger.info(str(e))
            return False
        finally:
            ser.close()
            logger.info("close serial")

    def checkStatus(self, ser):
        time.sleep(180)
        # print('wait 120 seconds')
        ren_cmd = 'ps -elf | grep render_service'
        laun_cmd = 'ps -elf | grep com.ohos.launcher'
        time.sleep(10)
        ren_status_1 = getPsPid(ser, ren_cmd, 'render_service')
        time.sleep(5)
        ren_status_2 = getPsPid(ser, ren_cmd, 'render_service')
        time.sleep(5)
        laun_status_1 = getPsPid(ser, laun_cmd, 'com.ohos.launcher')
        time.sleep(5)
        laun_status_2 = getPsPid(ser, laun_cmd, 'com.ohos.launcher')
        # wes_cmd = 'pidof weston'
        # laun_cmd = 'pidof com.ohos.launcher'
        # time.sleep(10)
        # wes_status_1 = sendCmd(ser, wes_cmd, READ_TIMEOUT)
        # time.sleep(5)
        # wes_status_2 = sendCmd(ser, wes_cmd, READ_TIMEOUT)
        # time.sleep(5)
        # laun_status_1 = sendCmd(ser, laun_cmd, READ_TIMEOUT)
        # time.sleep(5)
        # laun_status_2 = sendCmd(ser, laun_cmd, READ_TIMEOUT)
        logger.info('ren_status_1: %s ren_status_2: %s laun_1: %s laun_2: %s '%(ren_status_1,ren_status_2,laun_status_1,laun_status_2))
        if ren_status_1 and ren_status_1 == ren_status_2 and laun_status_1 and laun_status_1 == laun_status_2:
            return True
        if not ren_status_1 or not ren_status_2 or ren_status_1 != ren_status_2:
            # and not wes_status_1.strip().isdigit() or not wes_status_2.strip().isdigit()
            logger.printLog('process render_service is not exist')
        if not laun_status_1 or not laun_status_2 :
            # or not laun_status_1.strip().isdigit() or not laun_status_2.strip().isdigit()
            logger.printLog('process com.ohos.launcher is not exist')
        return False

    @timeout(30)
    def hdc_set_time(self,sn):
        # todo
        y_m_d = time.strftime('%Y-%m-%d',time.localtime())
        h_m_s = time.strftime('%H:%M:%S',time.localtime())
        cmd = 'hdc_std -t %s shell date -u "%sT%s"'% (sn,y_m_d,h_m_s)
        logger.printLog('hdc start set time')
        logger.printLog('cmd is %s'% cmd)
        ret,out = subprocess.getstatusoutput(cmd)
        logger.printLog('hdc end set time')
        if h_m_s not in out:
            logger.printLog('hdc设置时间失败')
            logger.printLog(out)
            return False
        logger.printLog(out)

    def eraseDevice(self, cmd, usb_port):
        '''
        ret 暂时没有作用，先使用out
        '''
        erase_retry = 0
        while erase_retry < 3:
            # 通电
            PowerOnByThread(usb_port)
            logger.info("cmd is: %s" % cmd)
            ret, outpri = subprocess.getstatusoutput(cmd)
            logger.info("flash fastboot result: %s " % ret)
            logger.info("print console: %s " % outpri)
            is_earse_again = any([True if item in outpri else False for item in error_str_list])
            if ret == 0 and erase_retry < 2 and not is_earse_again :
                logger.info('檫除成功'.center(20,'*'))
                break
            elif is_earse_again:
                logger.info('檫除存在问题 重新上下电 重新檫除')
                erase_retry += 1
                time.sleep(5)
                continue
            else:
                logger.info('other error')
                return False
        else:
            return False

    def catchEraseFail(self,cmd):
        logger.printLog(datetime.datetime.now())
        ret, outpri = subprocess.getstatusoutput(cmd)
        logger.printLog(datetime.datetime.now())
        return ret,outpri

def getPsPid(ser, cmd, ps):
    try:
        data = sendCmd(ser, cmd, READ_TIMEOUT)
        logger.info(data)
        data = [item for item in data.split('\n') if 'grep' not in item and ps in item ]
        if len(data) < 1:
            return
        elif len(data) > 1:
            data = max(data, key=len, default='')
        logger.info(data)
        data = data[0].split()
        if len(data) < 1:
            return
        else:
            logger.info(data)
            return data[1]
    except Exception as e :
        logger.info(str(e))
        logger.info('pid 获取失败')

def PowerOnByThread(usb_port,wait_time=10):
    thread = Thread(target=boardPowerOn, args=[usb_port, wait_time])
    thread.start()
    logger.info("thread board power on start")



def boardPowerOn(usb_port, waittime):
    logger.info("board power on start")
    time.sleep(waittime)

    #对端口下电
    if not usbPowerOnOff('127.0.0.1', '7788', usb_port, "off"):
        logger.error("board power off failed")
        return False

    #对端口上电
    if not usbPowerOnOff('127.0.0.1', '7788', usb_port, "on"):
        logger.error("board power on failed")
        return False
    logger.info("board power on end")


def getBoardType(ser):
    ret = sendCmd(ser, '\r', READ_TIMEOUT)
    if 'HMOS' in ret or 'OHOS' in ret or 'console:/ $' in ret or '#' in ret:
        ostype = 'OHOS'
    elif 'hisilicon' in ret:
        ostype = 'uboot'
    elif ' #' in ret:
        ostype = 'linux'
    else:
        ostype = 'bootrom'
    logger.info("board type is: %s" % ostype)
    return ostype

def sendCmd(ser, cmd, timeout):
    logger.info("cmd is: %s " % cmd)
    ser.write((cmd + '\n').encode())
    time.sleep(0.5)
    ret = ''
    i = 0
    while True:
        out = ser.read(ser.inWaiting())
        if not out:
            break
        if i > 12:
            break
        ret = ret + out.decode(encoding="utf-8", errors="ignore")
        time.sleep(timeout)
        i = i + 1
    logger.info("result is: %s " % ret)
    return ret

def sendHdcCmd(cmd):
    from subprocess import getstatusoutput
    i = 0
    while True:
        if i > 3:
            break
        try:
            out = executeHdcCmd(cmd)
            logger.info('hdc cmd success')
            return out
        except Exception as e:
            logger.info(str(e))
            logger.info('hdc cmd fail')
            i += 1
            time.sleep(1)
            continue
    logger.error('hdc cmd execute fail by three times')
    return ''

@timeout(10)
def executeHdcCmd(cmd):
    logger.info('cmd is %s' % cmd)
    time.sleep(0.5)
    ret, out = getstatusoutput(cmd)
    time.sleep(0.5)
    if ret != 0:
        logger.info('execute fail')
        return 'execute fail'
    logger.info('result is %s'% out)
    return out

if __name__ == '__main__':
    ser = serial.Serial('com5', 115200, timeout=0)