1#-*- coding:utf-8 -*- 2import uuid 3import sys 4import subprocess 5import os 6import serial 7 8from core.base import BaseApp, dec_stepmsg 9from util.file_locker import FileLock 10from util.log_info import logger 11from util.time_info import get_now_time_str_info, get_now_time_info, Timeout, timeout 12from aw.Telnet.TelnetClient import TelConnect 13from aw.Common.Constant import CONSTANT 14from aw.Download.Download import * 15from aw.Common.Common import getHostIp 16from aw.ExtractFile.ExtractFile import * 17from aw.poweronoff.serial_power_on_off import serialPowerOnOff 18 19lock_suffix = CONSTANT.File.LOCK_SUFFIX #通过文件锁实现并发下载 20suc_file = CONSTANT.File.SUC_FILE #通过本文件是区分版本是否成功下载 21failed_file = CONSTANT.File.FAILED_FILE #通过本文件是标记文件下载失败 22READ_MAXTIMEOUT = 600 23READ_TIMEOUT = 30 24READ_MINITIMEOUT = 5 25uboot_finish = 'hisilicon #' 26cmd_finish = ' #' 27 28class liteOsUpgrade_L3(BaseApp): 29 ''' 30 @author: w00278233 31 ''' 32 33 def __init__(self, param_file): 34 super().__init__(param_file) 35 self.param_List = ["upgrade_upgradeLocation"] 36 37 @dec_stepmsg("hongmeng L3 flash") 38 def excute(self): 39 ''' 40 #=================================================================================== 41 # @Method: excute(self) 42 # @Precondition: none 43 # @Func: 升级执行入口 44 # @PostStatus: none 45 # @eg: excute() 46 # @return: True or Flase 47 #=================================================================================== 48 ''' 49 step_index = self.params_dict.get("step_list").index("liteOsUpgrade_L3_app") 50 51 # 执行下载 52 try: 53 if not self.download(): 54 CONSTANT.ENVERRMESSAGE = "image download fail" 55 logger.printLog(CONSTANT.ENVERRMESSAGE) 56 return False 57 except Exception as e: 58 raise e 59 60 # 执行升级 61 try: 62 status = self.upgrade() 63 logger.info(status) 64 logger.info(type(status)) 65 logger.info(f'升级状态:{str(status)}') 66 if not status: 67 CONSTANT.ENVERRMESSAGE = "board upgrade fail" 68 logger.printLog(CONSTANT.ENVERRMESSAGE) 69 return False 70 return True 71 except Exception as e: 72 raise e 73 74 @dec_stepmsg("download") 75 @timeout(1800) 76 def download(self): 77 ''' 78 #=================================================================================== 79 # @Method: download(self) 80 # @Precondition: none 81 # @Func: 构建下载到本地的路径，执行相应包的下载 82 # @PostStatus: none 83 # @eg: download() 84 # @return: True or Flase 85 #=================================================================================== 86 ''' 87 global version_savepath, version_name 88 dir_path = CONSTANT.Path.getDirPath() 89 if self.params_dict.get("pbiid"): 90 version_path = self.params_dict.get("pbiid") 91 version_name = str(uuid.uuid5(uuid.NAMESPACE_URL, str(self.params_dict.get("pbiid")) + "FASTBOOT")) 92 version_savepath = os.path.join(dir_path, self.params_dict.get("flash_type"), version_name) 93 else: 94 version_path = self.params_dict.get("upgrade_upgradeLocation") 95 version_name = str(uuid.uuid5(uuid.NAMESPACE_URL, (self.params_dict.get("upgrade_upgradeLocation")))) 96 version_savepath = os.path.join(dir_path, version_name) 97 logger.printLog(version_savepath) 98 99 if self.params_dict.get("isDownload") == "True": 100 logger.printLog("不需要做下载，直接返回") 101 return True 102 103 #执行img下载 104 import hashlib 105 save_file_str = version_path.replace("/", "").replace("\\", "") 106 save_file_name = hashlib.sha1(save_file_str.encode("utf-8")).hexdigest() 107 logger.info("download hash string:%s, hash value:%s" % (save_file_str, save_file_name)) 108 save_path_file = os.path.join(dir_path, "record", "%s%s" % (save_file_name, ".txt")) 109 if not self.excutedown(version_path, os.path.join(version_savepath, "img"), save_path_file, False): 110 logger.error("download img fail") 111 return False 112 113 #保存本地版本路径给devicetest去版本路径下取用例 114 saveVersion(save_path_file, os.path.join(version_savepath, "img")) 115 116 # 执行升级脚本下载 117 if self.params_dict.get("upgrade_script"): 118 suc_mark = os.path.join(version_savepath, "scriptfile", suc_file) 119 if not self.excutedown(self.params_dict.get("upgrade_script"), 120 os.path.join(version_savepath, "scriptfile"), suc_mark, True): 121 logger.error("download upgrade script fail") 122 return False 123 with open(suc_mark, "a+") as fp: 124 fp.write("") 125 126 return True 127 128 def excutedown(self, source_path, download_dir, suc_mark, is_file): 129 ''' 130 #=================================================================================== 131 # @Method: excutedown(source_path, download_dir, is_file) 132 # @Precondition: none 133 # @Func: 执行下载动作 134 # @PostStatus: none 135 # @Param: source_path：资源文件路径 136 # download_dir:文件下载到本地的文件夹路径 137 # is_file:是否是文件 138 # 139 # @eg: excutedown("xxxx", "D:\\local\\image", Flase， os_method) 140 # @return: True or Flase 141 #=================================================================================== 142 ''' 143 failed_mark = os.path.join(download_dir, failed_file) 144 lock_path = os.path.join(download_dir, lock_suffix) 145 file_lock = FileLock() 146 if isDownLoadSuccess(download_dir, suc_mark, failed_mark): 147 return True 148 try: 149 nowtime = get_now_time_str_info() 150 logger.printLog("%s Downloading, please wait" % nowtime) 151 file_lock.lockFile(lock_path) 152 ret = "" 153 logger.info("Get lock. Start to ") 154 if self.params_dict.get("bt_enable") and self.params_dict.get("bt_enable") == "True": 155 ret = downloadByBitComet(source_path, download_dir, os_method) 156 elif source_path.startswith('\\\\'): 157 ret = downloadByCopy(source_path, download_dir, is_file) 158 elif self.params_dict.get("pbiid"): 159 ret = downlaodByDownloadTool(version_savepath, self.params_dict.get("version_type"), "FASTBOOT", self.params_dict.get("pbiid")) 160 elif source_path.startswith("http"): 161 # 临时修改下载链接 162 source_path = self.modifyHttpLink(source_path) 163 logger.info(f'modify path:{source_path}') 164 ret = downloadFileFromDevCloud(source_path, "", "", download_dir) 165 166 if source_path.endswith(".zip"): 167 zip_name = os.path.basename(source_path) 168 ret = extractZipFile(os.path.join(download_dir, zip_name), download_dir) 169 if source_path.endswith(".tar.gz") or (source_path.startswith("http") and ("file_id=" in source_path)): 170 if source_path.startswith("http") and ("file_id=" in source_path): 171 if source_path.endswith(".tar.gz"): 172 zip_name = source_path.split('=')[-1] 173 else: 174 zip_name = "out.tar.gz" 175 else: 176 zip_name = os.path.basename(source_path) 177 ret = unTarFile(os.path.join(download_dir, zip_name), download_dir) 178 nowtime = get_now_time_str_info() 179 logger.printLog("%s download to %s end" % (nowtime, download_dir)) 180 181 if not ret: 182 with open(failed_mark, "a+") as fp: 183 fp.write("") 184 return ret 185 except Exception as e: 186 logger.printLog(e) 187 raise Exception(e) 188 finally: 189 file_lock.releaseFile() 190 191 def modifyHttpLink(self,link): 192 try: 193 base_link = 'https://hm-verify.obs.cn-north-4.myhuaweicloud.com/' 194 filename = link.split('&')[-1].split('=')[-1] 195 names = filename.split('-') 196 link_1 = '/'.join(names[0:-1]) 197 link_2 = f'/{filename}' 198 final_link = base_link + link_1 + link_2 199 return final_link 200 except Exception as e: 201 logger.info('path is errror.check the link_path is right,please') 202 logger.info(str(e)) 203 204 205 @dec_stepmsg("upgrade") 206 @timeout(900) 207 def upgrade(self): 208 ''' 209 #=================================================================================== 210 # @Method: upgrade(self) 211 # @Precondition: none 212 # @Func: 升级相关业务逻辑 213 # @PostStatus: none 214 # @eg: upgrade() 215 # @return: True or Flase 216 #=================================================================================== 217 ''' 218 try: 219 #time.sleep(1000) 220 logger.printLog('开始升级') 221 self.enterTheDevice() 222 time.sleep(5) 223 self.flashSystemImg() 224 time.sleep(5) 225 self.rebootDevice() 226 return True 227 except Exception as e: 228 logger.error('***********************************') 229 logger.error(str(e)) 230 logger.error('flash fail') 231 logger.error('***********************************') 232 return False 233 234 #@timeout(30*3) 235 def enterTheDevice(self): 236 logger.info('enter fastboot status'.center(50,'*')) 237 devices_status_count = 0 238 while devices_status_count < 3 : 239 if self.isFastbootStatus(): 240 break 241 devices_status_count += 1 242 time.sleep(20) 243 else: 244 logger.error(f'device {self.params_dict.get("sn")} is not exist ,please make sure the sn is True') 245 raise Exception('ERROR_SN_DEVICE_NOT_EXIST') 246 247 #@timeout(20) 248 def isFastbootStatus(self): 249 adb_retcode, adb_output = subprocess.getstatusoutput('adb devices') 250 251 fastboot_retcode, fastboot_output = subprocess.getstatusoutput('fastboot devices') 252 253 if self.params_dict.get('sn') in adb_output: 254 retcode, output = subprocess.getstatusoutput('adb -s %s reboot fastboot' % self.params_dict.get('sn')) 255 logger.info('adb -s %s reboot fastboot retcode: %s'% (self.params_dict.get('sn'),str(retcode))) 256 logger.info('adb -s %s reboot fastboot output: %s'% (self.params_dict.get('sn'),output)) 257 logger.info('device from adb mode to fastboot mode,please wait') 258 time.sleep(30) 259 retcode, output = subprocess.getstatusoutput('fastboot devices') 260 logger.info('fastboot devices fastboot_retcode: %s'% str(fastboot_retcode)) 261 logger.info('fastboot devices fastboot_output: %s'% fastboot_output) 262 if self.params_dict.get('sn') in output: 263 logger.info('device enter fastboot mode success') 264 return True 265 else: 266 logger.info('device enter fastboot mode fail') 267 return False 268 elif self.params_dict.get('sn') in fastboot_output: 269 logger.info('device has been in fastboot mode') 270 return True 271 else: 272 logger.info('设备未找到，重新尝试') 273 return False 274 275 @timeout(60*2*3) 276 def flashSystemImg(self): 277 logger.info('flash system.img '.center(50,'*')) 278 flash_count = 0 279 while flash_count < 3: 280 status = self.isFlashSuccess() 281 logger.info(status) 282 if status: 283 break 284 flash_count +=1 285 time.sleep(3) 286 else: 287 logger.error('system.img镜像刷写失败') 288 raise Exception('[ERROR] ERROR_FLASH_SYSTEM.IMG_FAIL') 289 290 @timeout(60*2) 291 def isFlashSuccess(self): 292 #local_image_path = os.path.join(version_savepath, 'img', r'aosp\target\product\generic_arm64', 'system.img') 293 local_image_path = os.path.join(version_savepath, 'img', 'system.img') 294 logger.info(f'flash img path: {local_image_path}') 295 if os.path.exists(local_image_path): 296 retcode, output = subprocess.getstatusoutput('fastboot -s %s flash system %s' % (self.params_dict.get('sn'), local_image_path)) 297 logger.info('fastboot -s %s flash system %s retcode: %s'% (self.params_dict.get('sn'),local_image_path,str(retcode))) 298 logger.info('fastboot -s %s flash system %s output: %s'% (self.params_dict.get('sn'),local_image_path,output)) 299 if retcode == 0: 300 logger.info('flash system.img success') 301 return True 302 else: 303 flash_count += 1 304 logger.info('再次尝试刷system.img镜像') 305 306 else: 307 logger.error('镜像文件路径不对') 308 raise Exception('[ERROR] ERROR_IMG_PATH_NOT_EXIST') 309 310 def rebootDevice(self): 311 logger.info('reboot the device '.center(50,'*')) 312 reboot_count = 0 313 while reboot_count < 3 : 314 logger.info('start reboot') 315 if self.isRebootSuccess(): 316 break 317 else: 318 reboot_count +=1 319 time.sleep(10) 320 else: 321 raise Exception('[ERROR] ERROR_REBOOT_DEVICE_FAIL') 322 323 def isRebootSuccess(self): 324 retcode, output = subprocess.getstatusoutput('fastboot -s %s reboot' % self.params_dict.get('sn')) 325 logger.info('fastboot -s %s reboot retcode: %s'% (self.params_dict.get('sn'),str(retcode))) 326 logger.info('fastboot -s %s reboot output: %s'% (self.params_dict.get('sn'),output)) 327 time.sleep(50) 328 retcode, output = subprocess.getstatusoutput('adb devices') 329 logger.info('adb devices: %s'% str(retcode)) 330 logger.info('adb devices: %s'% output) 331 if self.params_dict.get('sn') in output: 332 logger.info('reboot success') 333 return True 334 else: 335 logger.info('reboot fail') 336 return False 337 338 339 340if __name__ == '__main__': 341 liteOsUpgrade_L3.modifyHttpLink(liteOsUpgrade_L3, 342 'https://devrepo.devcloud.cn-north-4.huaweicloud.com/DevRepoServer/v1/files/download?file_id=0fe39c430a364f59bd040c53e7a3ef1c&type=archive&filename=Daily_Version#2021-05-19_06-38-31#L35_aosp_arm64.tar.gz')