mirror of
https://github.com/openharmony/developtools_integration_verification.git
synced 2026-07-24 13:45:54 -04:00
openharmony_ci
Regular → Executable
@@ -1,498 +0,0 @@
|
||||
#-*- coding:utf-8 -*-
|
||||
import uuid
|
||||
import sys
|
||||
import subprocess
|
||||
import os
|
||||
import serial
|
||||
|
||||
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
|
||||
|
||||
lock_suffix = CONSTANT.File.LOCK_SUFFIX #通过文件锁实现并发下载
|
||||
suc_file = CONSTANT.File.SUC_FILE #通过本文件是区分版本是否成功下载
|
||||
failed_file = CONSTANT.File.FAILED_FILE #通过本文件是标记文件下载失败
|
||||
READ_MAXTIMEOUT = 120
|
||||
READ_TIMEOUT = 30
|
||||
READ_MINITIMEOUT = 5
|
||||
uboot_finish = 'hisilicon #'
|
||||
cmd_finish = ' #'
|
||||
|
||||
class liteOsUpgrade_L1_shequ_dv(BaseApp):
|
||||
'''
|
||||
@author: w00278233
|
||||
'''
|
||||
|
||||
def __init__(self, param_file):
|
||||
super().__init__(param_file)
|
||||
self.param_List = ["deploy_com",
|
||||
"usb_port",
|
||||
"upgrade_upgradeLocation"]
|
||||
|
||||
@dec_stepmsg("hongmeng L1 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_L1_shequ_dv_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(1800)
|
||||
def download(self):
|
||||
'''
|
||||
#===================================================================================
|
||||
# @Method: download(self)
|
||||
# @Precondition: none
|
||||
# @Func: 构建下载到本地的路径，执行相应包的下载
|
||||
# @PostStatus: none
|
||||
# @eg: download()
|
||||
# @return: True or Flase
|
||||
#===================================================================================
|
||||
'''
|
||||
global version_savepath, version_name
|
||||
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 string:%s, hash value:%s" % (save_file_str, save_file_name))
|
||||
save_path_file = os.path.join(dir_path, "record", "%s%s" % (save_file_name, ".txt"))
|
||||
if not self.excutedown(version_path, os.path.join(version_savepath, "img"), save_path_file, False):
|
||||
logger.error("download img fail")
|
||||
return False
|
||||
|
||||
#保存本地版本路径给devicetest去版本路径下取用例
|
||||
saveVersion(save_path_file, os.path.join(version_savepath, "img"))
|
||||
|
||||
#将本地版本push到远程tftp服务器
|
||||
tftp_ip = self.params_dict.get("Tftpserver_IP")
|
||||
if tftp_ip:
|
||||
box_ip = subprocess.getoutput("ipconfig")
|
||||
logger.info("current box ip is: %s" % box_ip)
|
||||
if not tftp_ip in box_ip:
|
||||
logger.info("start to send file")
|
||||
return self.pushFileToTftpServer(tftp_ip, os.path.join(version_savepath, "img"), version_name)
|
||||
return True
|
||||
|
||||
def pushFileToTftpServer(self, tftp_ip, local_file_path, dis_path, username="root", passwd="devicetest@2012"):
|
||||
push_suc = os.path.join(local_file_path, "push_succ.txt")
|
||||
push_fail = os.path.join(local_file_path, "push_fail.txt")
|
||||
push_lock = os.path.join(local_file_path, "push.lock")
|
||||
if os.path.isfile(push_suc):
|
||||
logger.info("push file to tftpserver by other process.")
|
||||
return True
|
||||
elif os.path.isfile(push_fail):
|
||||
logger.info("%s exists. Other process has push package and failed. It will push by self" % push_fail)
|
||||
os.remove(push_fail)
|
||||
else:
|
||||
if os.path.isfile(push_lock):
|
||||
logger.info("wait for other process send succ")
|
||||
ret = waitotherdown(push_suc)
|
||||
return ret
|
||||
|
||||
file_lock = FileLock()
|
||||
file_lock.lockFile(push_lock)
|
||||
|
||||
try:
|
||||
import paramiko
|
||||
logger.info("ip: %s, source_path: %s" % (tftp_ip, local_file_path))
|
||||
#client = paramiko.Transport((tftp_ip,22), default_window_size=3072)
|
||||
client = paramiko.Transport((tftp_ip,22))
|
||||
client.connect(username=username,password=passwd)
|
||||
sftp = paramiko.SFTPClient.from_transport(client)
|
||||
#检查tftp服务器根目录，/data或/data/data
|
||||
ret = sftp.listdir(r"/data")
|
||||
server_base_dir = "/data/MobileUpgrade/local_img"
|
||||
if "data" in ret:
|
||||
server_base_dir = "/data/data/MobileUpgrade/local_img"
|
||||
logger.info("server base dir is : %s " % server_base_dir)
|
||||
#检查tftp服务器上有没有本次的版本目录，如果没有则创建文件夹
|
||||
ret = sftp.listdir(server_base_dir)
|
||||
server_des_path = "%s/%s" % (server_base_dir, dis_path)
|
||||
if not dis_path in ret:
|
||||
logger.info("mkdir in tftpserver: %s " % server_des_path)
|
||||
sftp.mkdir(server_des_path)
|
||||
server_img_path = "%s/%s/img" % (server_base_dir, dis_path)
|
||||
logger.info("mkdir in tftpserver: %s " % server_img_path)
|
||||
sftp.mkdir(server_img_path)
|
||||
remote_path = "%s/%s/img" % (server_base_dir, dis_path)
|
||||
logger.info("%s exists." % remote_path)
|
||||
ret = sftp.listdir(remote_path)
|
||||
#如果tftp服务器上有push_succ文件夹，则说明已经由其它进程或box把版本上传成功了。
|
||||
if "push_succ" in ret:
|
||||
logger.info("send file succ by other process")
|
||||
with open(push_suc, "a+") as fp:
|
||||
fp.write("")
|
||||
return True
|
||||
#如果tftp服务器上有push.lock文件夹，则说明有其它进程在向tftp服务器发送版本，等待其它进程发送完毕。
|
||||
elif "push.lock" in ret:
|
||||
logger.info("wait for other process")
|
||||
i = 0
|
||||
while True:
|
||||
time.sleep(10)
|
||||
ret = sftp.listdir(remote_path)
|
||||
if "push_succ" in ret:
|
||||
logger.info("other process succ")
|
||||
return True
|
||||
if i > 6:
|
||||
logger.info("wait 60s, other process failed")
|
||||
break
|
||||
i = i + 1
|
||||
#由本进程向tftp服务器发送版本，先创建push.lock占用上传资源
|
||||
logger.info("mkdir in tftpserver: %s/push.lock " % remote_path)
|
||||
sftp.mkdir("%s/push.lock" % remote_path)
|
||||
pushfilelist = ["OHOS_Image.bin", "rootfs.img", "userfs.img", "rootfs_vfat.img", "userfs_vfat.img", "rootfs_jffs2.img", "userfs_jffs2.img"]
|
||||
for filename in pushfilelist:
|
||||
if not os.path.exists(os.path.join(local_file_path, filename)):
|
||||
continue
|
||||
logger.info("copy %s to %s" % (os.path.join(local_file_path, filename), "%s/%s" % (remote_path, filename)))
|
||||
try:
|
||||
sftp.put(os.path.join(local_file_path, filename), "%s/%s" % (remote_path, filename))
|
||||
except Exception as e:
|
||||
logger.info("retry, copy %s to %s" % (os.path.join(local_file_path, filename), "%s/%s" % (remote_path, filename)))
|
||||
sftp.put(os.path.join(local_file_path, filename), "%s/%s" % (remote_path, filename))
|
||||
#创建上传成功标识
|
||||
logger.info("mkdir in tftpserver: %s/push_succ " % remote_path)
|
||||
sftp.mkdir("%s/push_succ" % remote_path)
|
||||
#删除占用标识
|
||||
logger.info("del %s/push.lock " % remote_path)
|
||||
sftp.rmdir("%s/push.lock" % remote_path)
|
||||
with open(push_suc, "a+") as fp:
|
||||
fp.write("")
|
||||
time.sleep(10)
|
||||
return True
|
||||
except (OSError, Exception) as exception:
|
||||
logger.error("copy to file to tftp server failed with error {}".format(exception))
|
||||
logger.info("del %s/%s " % (server_base_dir, dis_path))
|
||||
sftp.rmdir("%s/%s" % (server_base_dir, dis_path))
|
||||
client.close()
|
||||
with open(push_fail, "a+") as fp:
|
||||
fp.write("")
|
||||
return False
|
||||
finally:
|
||||
logger.info("close client")
|
||||
client.close()
|
||||
|
||||
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)
|
||||
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
|
||||
#===================================================================================
|
||||
'''
|
||||
|
||||
deploy_com = self.params_dict.get("deploy_com")
|
||||
usb_port = self.params_dict.get("usb_port")
|
||||
baudrate = self.params_dict.get("baudrate")
|
||||
tftp_ip = self.params_dict.get("Tftpserver_IP")
|
||||
device_ip = self.params_dict.get("Device_IP")
|
||||
device_Mac = self.params_dict.get("Device_MAC")
|
||||
device_netmask = self.params_dict.get("Device_Netmask")
|
||||
device_gatewayip = self.params_dict.get("Device_GatewayIP")
|
||||
#芯片类型,根据芯片类型获取对应的刷机命令
|
||||
flash_type = self.params_dict.get("flash_type")
|
||||
|
||||
if not deploy_com:
|
||||
logger.error("deploy_com is NULL !!")
|
||||
return False
|
||||
if not baudrate:
|
||||
baudrate = 115200
|
||||
if not device_netmask:
|
||||
device_netmask = "255.255.252.0"
|
||||
if not device_Mac:
|
||||
device_Mac = "3a:82:d0:08:f4:99"
|
||||
local_image_path = "%s/%s" % (version_name, "img")
|
||||
scriptpath = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(os.path.dirname(__file__)))))
|
||||
scriptfile = os.path.join(scriptpath, "resource", "L1", "dv300_shequ", "update.txt")
|
||||
logger.info("upgrade scriptfile is: %s" % scriptfile)
|
||||
if not os.path.exists(scriptfile):
|
||||
logger.error("%s is not exit" % scriptfile)
|
||||
return False
|
||||
|
||||
logger.info("open com")
|
||||
ser = serial.Serial(deploy_com, int(baudrate), timeout=1)
|
||||
try:
|
||||
logger.info("打开serial")
|
||||
if ser.is_open == False:
|
||||
ser.open()
|
||||
|
||||
logger.info("获取单板状态")
|
||||
board_type = getBoardType(ser)
|
||||
if board_type != "uboot":
|
||||
sendResetCmd(ser, READ_TIMEOUT)
|
||||
board_type = getBoardType(ser)
|
||||
if board_type == "bootrom":
|
||||
if not recoveryBoard(ser, usb_port):
|
||||
logger.error("recovery board fail, please check board")
|
||||
return False
|
||||
|
||||
with open(scriptfile, "r") as fp:
|
||||
lines = fp.readlines()
|
||||
for line in lines:
|
||||
#line = line.strip()
|
||||
if not line:
|
||||
logger.info("cmd is: %s " % line)
|
||||
continue
|
||||
if "%Tftpserver_IP%" in line:
|
||||
line = line.replace("%Tftpserver_IP%", tftp_ip)
|
||||
if "%Device_IP%" in line:
|
||||
line = line.replace("%Device_IP%", device_ip)
|
||||
if "%Device_MAC%" in line:
|
||||
line = line.replace("%Device_MAC%", device_Mac)
|
||||
if "%Device_Netmask%" in line:
|
||||
line = line.replace("%Device_Netmask%", device_netmask)
|
||||
if "%Device_GatewayIP%" in line:
|
||||
line = line.replace("%Device_GatewayIP%", device_gatewayip)
|
||||
if "userfs.img" in line:
|
||||
if not os.path.exists(os.path.join(version_savepath, "img", "userfs.img")):
|
||||
line = line.replace("userfs.img", "userfs_vfat.img")
|
||||
if "rootfs.img" in line:
|
||||
if not os.path.exists(os.path.join(version_savepath, "img", "rootfs.img")):
|
||||
line = line.replace("rootfs.img", "rootfs_vfat.img")
|
||||
if "tftp" in line:
|
||||
packagefile_path = "/%s" % flash_type
|
||||
new_packagefile_path = "/%s" % local_image_path
|
||||
line = line.replace(packagefile_path, new_packagefile_path)
|
||||
ret = sendCmd(ser, line, "hisilicon #", READ_MAXTIMEOUT)
|
||||
if "retry count exceeded; starting again" in ret.lower():
|
||||
logger.info("tftp cmd retry again!!")
|
||||
ret = sendCmd(ser, line, "hisilicon #", READ_MAXTIMEOUT)
|
||||
if "error:" in ret.lower():
|
||||
logger.error("tftp fail")
|
||||
return False
|
||||
continue
|
||||
if "reset" in line:
|
||||
ret = sendCmd(ser, line, "OHOS #", READ_TIMEOUT)
|
||||
continue
|
||||
ret = sendCmd(ser, line, " #", READ_MINITIMEOUT)
|
||||
board_type = getBoardType(ser)
|
||||
if board_type == "OHOS":
|
||||
logger.info("upgrade success")
|
||||
init_cmd = "ifconfig eth0 %s netmask %s gateway %s \r" % (device_ip, device_netmask, device_gatewayip)
|
||||
sendCmd(ser, init_cmd, "OHOS #", READ_MINITIMEOUT)
|
||||
sendCmd(ser, 'ifconfig\r', "OHOS #", READ_MINITIMEOUT)
|
||||
return True
|
||||
else:
|
||||
logger.info("upgrade fail")
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.printLog(e)
|
||||
return False
|
||||
finally:
|
||||
ser.close()
|
||||
logger.info("close serial")
|
||||
|
||||
|
||||
def boardPowerOn(usb_port):
|
||||
logger.info("board power on start")
|
||||
|
||||
#对端口下电
|
||||
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_MINITIMEOUT)
|
||||
if 'HMOS' in ret or 'OHOS' 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, endtag, timeout):
|
||||
ser.reset_input_buffer()
|
||||
logger.info("cmd is: %s " % cmd)
|
||||
ser.write((cmd + '\n').encode())
|
||||
time.sleep(0.5)
|
||||
i = 0
|
||||
ret = ''
|
||||
while True:
|
||||
out = ser.read(ser.inWaiting())
|
||||
ret = ret + out.decode(encoding="utf-8", errors="ignore")
|
||||
if endtag in ret:
|
||||
logger.info("cmd end")
|
||||
break
|
||||
time.sleep(1)
|
||||
i = i + 1
|
||||
if i > timeout:
|
||||
logger.info("cmd timeout")
|
||||
break
|
||||
logger.info("result is: %s " % ret)
|
||||
return ret
|
||||
|
||||
def sendResetCmd(ser, timeout):
|
||||
logger.info("send reset")
|
||||
ser.reset_input_buffer()
|
||||
ser.write(b"reset \r\n")
|
||||
ret = ser.read(ser.inWaiting())
|
||||
time.sleep(3)
|
||||
#logger.info("result is: %s " % ret)
|
||||
ser.write(b"\r\n")
|
||||
ret = ser.read(ser.inWaiting())
|
||||
logger.info("result is: %s " % ret)
|
||||
ser.write(b"\r\n")
|
||||
ret = ser.read(ser.inWaiting())
|
||||
logger.info("result is: %s " % ret)
|
||||
ser.write(b"\r\n")
|
||||
ret = ser.read(ser.inWaiting())
|
||||
logger.info("result is: %s " % ret)
|
||||
return ret
|
||||
|
||||
|
||||
def recoveryBoard(ser, usb_port):
|
||||
logger.info("start recovery board")
|
||||
boardPowerOn(usb_port)
|
||||
#再发送回车
|
||||
ser.write(b"\r\n")
|
||||
time.sleep(0.5)
|
||||
ret = ser.read(ser.inWaiting())
|
||||
logger.info("result is: %s " % ret)
|
||||
ser.write(b"\r\n")
|
||||
time.sleep(0.5)
|
||||
ret = ser.read(ser.inWaiting())
|
||||
logger.info("result is: %s " % ret)
|
||||
board_type = getBoardType(ser)
|
||||
if board_type == "bootrom":
|
||||
logger.error("recovery board failed")
|
||||
return False
|
||||
logger.info("recovery board succ")
|
||||
return True
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
param_file = sys.argv[1]
|
||||
if not param_file:
|
||||
logger.printLog("Missing params file")
|
||||
sys.exit(-1)
|
||||
try:
|
||||
uphandle = liteOsUpgrade_L1(param_file)
|
||||
uphandle._excuteApp()
|
||||
except Exception as e:
|
||||
logger.printLog(e)
|
||||
sys.exit(-1)
|
||||
@@ -1,443 +0,0 @@
|
||||
#-*- coding:utf-8 -*-
|
||||
import uuid
|
||||
import sys
|
||||
import subprocess
|
||||
import os
|
||||
import serial
|
||||
|
||||
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
|
||||
|
||||
lock_suffix = CONSTANT.File.LOCK_SUFFIX #通过文件锁实现并发下载
|
||||
suc_file = CONSTANT.File.SUC_FILE #通过本文件是区分版本是否成功下载
|
||||
failed_file = CONSTANT.File.FAILED_FILE #通过本文件是标记文件下载失败
|
||||
READ_MAXTIMEOUT = 60
|
||||
READ_TIMEOUT = 30
|
||||
READ_MINITIMEOUT = 5
|
||||
uboot_finish = 'hisilicon #'
|
||||
cmd_finish = ' #'
|
||||
|
||||
class liteOsUpgrade_L1_shequ_test(BaseApp):
|
||||
'''
|
||||
@author: w00278233
|
||||
'''
|
||||
|
||||
def __init__(self, param_file):
|
||||
super().__init__(param_file)
|
||||
self.param_List = ["deploy_com",
|
||||
"usb_port",
|
||||
"upgrade_upgradeLocation"]
|
||||
|
||||
@dec_stepmsg("hongmeng L1 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_L1_shequ_test_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(1800)
|
||||
def download(self):
|
||||
'''
|
||||
#===================================================================================
|
||||
# @Method: download(self)
|
||||
# @Precondition: none
|
||||
# @Func: 构建下载到本地的路径，执行相应包的下载
|
||||
# @PostStatus: none
|
||||
# @eg: download()
|
||||
# @return: True or Flase
|
||||
#===================================================================================
|
||||
'''
|
||||
global version_savepath, version_name
|
||||
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"))
|
||||
if not self.excutedown(version_path, os.path.join(version_savepath, "img"), save_path_file, False):
|
||||
logger.error("download img fail")
|
||||
return False
|
||||
|
||||
#保存本地版本路径给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)
|
||||
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
|
||||
#===================================================================================
|
||||
'''
|
||||
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
|
||||
|
||||
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
|
||||
scriptpath = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(os.path.dirname(__file__)))))
|
||||
#升级需要的工具归档
|
||||
logger.printLog('归档工具')
|
||||
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
|
||||
#将升级需要的文件拷贝到镜像里面
|
||||
logger.printLog('拷贝镜像')
|
||||
local_image_path = os.path.join(version_savepath, "img")
|
||||
old_xml_path = os.path.join(scriptpath, "resource", "L1", flash_type, "usb-burn.xml")
|
||||
xml_path = os.path.join(local_image_path, "usb-burn.xml")
|
||||
if chip_version == 'hi3518':
|
||||
old_xml_path = os.path.join(scriptpath, "resource", "L1", flash_type, "usb-burn-jffs2.xml")
|
||||
elif chip_version == 'hi3516':
|
||||
old_xml_path = os.path.join(scriptpath, "resource", "L1", flash_type, "usb-burn-vfat.xml")
|
||||
copyFile(old_xml_path, xml_path)
|
||||
if flash_type.lower() == "ev300":
|
||||
chip_type = "Hi3518EV300"
|
||||
ubootpath = os.path.join(scriptpath, "resource", "L1", flash_type.lower(), "u-boot-hi3518ev300.bin")
|
||||
elif flash_type.lower() == "dv300":
|
||||
chip_type = "Hi3516DV300"
|
||||
ubootpath = os.path.join(scriptpath, "resource", "L1", flash_type.lower(), "u-boot-hi3516dv300.bin")
|
||||
else:
|
||||
logger.error("flash_type is : %s " % flash_type)
|
||||
return False
|
||||
copyFile(ubootpath, local_image_path)
|
||||
scriptfile = os.path.join(scriptpath, "resource", "L1", flash_type.lower(), "update.txt")
|
||||
|
||||
from threading import Thread
|
||||
thread = Thread(target=boardPowerOn, args=[usb_port, 10])
|
||||
thread.start()
|
||||
|
||||
current_path = os.getcwd()
|
||||
logger.info("before excute hiburn,current path is: %s" % current_path)
|
||||
os.chdir(hiburntoolpath)
|
||||
logger.info("excute hiburn path is: %s" % os.getcwd())
|
||||
#擦除fastboot
|
||||
logger.printLog('擦除fastboot')
|
||||
flash_uboot_xml = os.path.join(scriptpath, "resource", "L1", 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)
|
||||
logger.info("cmd is: %s" % cmd)
|
||||
ret, outpri = subprocess.getstatusoutput(cmd)
|
||||
logger.info("flash fastboot result: %s " % ret)
|
||||
logger.info("print console: %s " % outpri)
|
||||
|
||||
retry = 0
|
||||
while retry < 2:
|
||||
|
||||
#usb刷机
|
||||
logger.printLog('开始刷机')
|
||||
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:
|
||||
time.sleep(10)
|
||||
retry = retry + 1
|
||||
logger.printLog('刷机未成功再刷一次')
|
||||
continue
|
||||
logger.printLog(ret)
|
||||
logger.error("hiburn usb upgrade failed!!")
|
||||
return False
|
||||
retry = retry + 3
|
||||
os.chdir(current_path)
|
||||
logger.info("hiburn upgrade end, check board status")
|
||||
time.sleep(5)
|
||||
ser = serial.Serial(deploy_com, int(baudrate), timeout=1)
|
||||
reset_count = 0
|
||||
try:
|
||||
while reset_count < 2:
|
||||
logger.info("打开serial")
|
||||
if ser.is_open == False:
|
||||
ser.open()
|
||||
|
||||
logger.info("获取单板状态")
|
||||
board_type = getBoardType(ser)
|
||||
logger.printLog('before reset')
|
||||
logger.printLog(f'board_type:{board_type}')
|
||||
if board_type == "uboot":
|
||||
with open(scriptfile, "r") as fp:
|
||||
lines = fp.readlines()
|
||||
for line in lines:
|
||||
if not line:
|
||||
logger.info("cmd is: %s " % line)
|
||||
continue
|
||||
if "reset" in line:
|
||||
ret = sendCmd(ser, line, READ_MAXTIMEOUT)
|
||||
continue
|
||||
ret = sendCmd(ser, line, READ_MINITIMEOUT)
|
||||
board_type = getBoardType(ser)
|
||||
logger.printLog('after reset')
|
||||
logger.printLog(f'board_type:{board_type}')
|
||||
|
||||
if True:
|
||||
reset_count += 1
|
||||
logger.printLog('再次重启')
|
||||
continue
|
||||
time.sleep(1000)
|
||||
if flash_type.lower() == "dv300":
|
||||
logger.info("upgrade success")
|
||||
init_cmd = "ifconfig eth0 %s netmask %s gateway %s \r" % (device_ip, device_netmask, device_gatewayip)
|
||||
sendCmd(ser, init_cmd, READ_MINITIMEOUT)
|
||||
sendCmd(ser, 'ifconfig\r', READ_MINITIMEOUT)
|
||||
return True
|
||||
elif flash_type.lower() == "ev300":
|
||||
logger.info("setup wifi")
|
||||
cmd = 'ls\r'
|
||||
ret = sendCmd(ser, cmd, READ_MINITIMEOUT)
|
||||
if not "sdcard" in ret.lower():
|
||||
cmd = 'mkdir /sdcard\r'
|
||||
ret = sendCmd(ser, cmd, READ_MINITIMEOUT)
|
||||
if "error:" in ret.lower():
|
||||
logger.error("mkdir /sdcard fail")
|
||||
return False
|
||||
cmd = 'mount /dev/mmcblk0p0 /sdcard vfat\r'
|
||||
ret = sendCmd(ser, cmd, READ_MINITIMEOUT)
|
||||
cmd = 'ls /sdcard\r'
|
||||
ret = sendCmd(ser, cmd, READ_MINITIMEOUT)
|
||||
cmd = 'cd /sdcard/wpa\r'
|
||||
ret = sendCmd(ser, cmd, READ_MINITIMEOUT)
|
||||
cmd = 'exec wpa_supplicant -i wlan0 -c wpa_supplicant.conf \r'
|
||||
ret = sendCmd(ser, cmd, READ_MAXTIMEOUT)
|
||||
if "error:" in ret.lower():
|
||||
logger.error("setup wifi fail")
|
||||
return False
|
||||
cmd = 'ifconfig\r'
|
||||
ret = sendCmd(ser, cmd, READ_MINITIMEOUT)
|
||||
if "error:" in ret.lower():
|
||||
logger.error("ifconfig fail")
|
||||
return False
|
||||
logger.info("upgrade success")
|
||||
return True
|
||||
else:
|
||||
logger.error("upgrade fail")
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.printLog(e)
|
||||
return False
|
||||
finally:
|
||||
ser.close()
|
||||
logger.info("close serial")
|
||||
|
||||
|
||||
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:
|
||||
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 > 2:
|
||||
break
|
||||
ret = ret + out.decode(encoding="utf-8", errors="ignore")
|
||||
time.sleep(timeout)
|
||||
i = i + 1
|
||||
logger.info("result is: %s " % ret)
|
||||
return ret
|
||||
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
param_file = sys.argv[1]
|
||||
if not param_file:
|
||||
logger.printLog("Missing params file")
|
||||
sys.exit(-1)
|
||||
try:
|
||||
uphandle = liteOsUpgrade_L1(param_file)
|
||||
uphandle._excuteApp()
|
||||
except Exception as e:
|
||||
logger.printLog(e)
|
||||
sys.exit(-1)
|
||||
@@ -1,364 +0,0 @@
|
||||
#-*- coding:utf-8 -*-
|
||||
import uuid
|
||||
import sys
|
||||
import subprocess
|
||||
import os
|
||||
import serial
|
||||
|
||||
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
|
||||
from threading import Thread
|
||||
|
||||
lock_suffix = CONSTANT.File.LOCK_SUFFIX #通过文件锁实现并发下载
|
||||
suc_file = CONSTANT.File.SUC_FILE #通过本文件是区分版本是否成功下载
|
||||
failed_file = CONSTANT.File.FAILED_FILE #通过本文件是标记文件下载失败
|
||||
READ_MAXTIMEOUT = 20
|
||||
READ_TIMEOUT = 30
|
||||
READ_MINITIMEOUT = 5
|
||||
uboot_finish = 'hisilicon #'
|
||||
cmd_finish = ' #'
|
||||
|
||||
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
|
||||
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
|
||||
|
||||
#保存本地版本路径给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
|
||||
#===================================================================================
|
||||
'''
|
||||
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
|
||||
|
||||
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
|
||||
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
|
||||
#将升级需要的文件拷贝到镜像里面
|
||||
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)
|
||||
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
|
||||
# copyFile(ubootpath, local_image_path)
|
||||
scriptfile = os.path.join(scriptpath, "resource", "L2", f'{flash_type.lower()}', "update.txt")
|
||||
logger.info(f'scriptfile:{scriptfile}')
|
||||
|
||||
current_path = os.getcwd()
|
||||
logger.info("before excute hiburn,current path is: %s" % current_path)
|
||||
os.chdir(hiburntoolpath)
|
||||
logger.info("excute hiburn path is: %s" % os.getcwd())
|
||||
#擦除fastboot
|
||||
logger.printLog('erase fastboot')
|
||||
flash_uboot_xml = os.path.join(scriptpath, "resource", "L2", flash_type.lower(), "flash_fastboot.xml")
|
||||
erase_retry = 0
|
||||
while erase_retry < 3:
|
||||
# 通电
|
||||
PowerOnByThread(usb_port)
|
||||
cmd = ".\jre\\bin\java -jar hiburn.jar --erase -n %s -m serial %s -x %s" % (chip_type, deploy_com.upper(), flash_uboot_xml)
|
||||
logger.info("cmd is: %s" % cmd)
|
||||
ret, outpri = subprocess.getstatusoutput(cmd)
|
||||
logger.info("flash fastboot result: %s " % ret)
|
||||
logger.info("print console: %s " % outpri)
|
||||
if ret == 0 and 'Unknown' not in outpri and erase_retry < 2:
|
||||
break
|
||||
elif 'Unknown' in outpri:
|
||||
logger.info('串口问题 重新上下电 重新檫除')
|
||||
erase_retry += 1
|
||||
time.sleep(5)
|
||||
continue
|
||||
else:
|
||||
logger.info('other error')
|
||||
return False
|
||||
else:
|
||||
return False
|
||||
|
||||
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")
|
||||
return True
|
||||
|
||||
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:
|
||||
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 > 2:
|
||||
break
|
||||
ret = ret + out.decode(encoding="utf-8", errors="ignore")
|
||||
time.sleep(timeout)
|
||||
i = i + 1
|
||||
logger.info("result is: %s " % ret)
|
||||
return ret
|
||||
|
||||
|
||||
Regular → Executable
Reference in New Issue
Block a user