# Copyright (c) Quectel Wireless Solution, Co., Ltd.All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import uos import fota import app_fota import uzlib import ql_fs import uhashlib import ubinascii import app_fota_download from queue import Queue from usr.logging import getLogger from usr.settings import SYSNAME from usr.settings import PROJECT_NAME log = getLogger(__name__) FOTA_ERROR_CODE = { 1001: "FOTA_DOMAIN_NOT_EXIST", 1002: "FOTA_DOMAIN_TIMEOUT", 1003: "FOTA_DOMAIN_UNKNOWN", 1004: "FOTA_SERVER_CONN_FAIL", 1005: "FOTA_AUTH_FAILED", 1006: "FOTA_FILE_NOT_EXIST", 1007: "FOTA_FILE_SIZE_INVALID", 1008: "FOTA_FILE_GET_ERR", 1009: "FOTA_FILE_CHECK_ERR", 1010: "FOTA_INTERNAL_ERR", 1011: "FOTA_NOT_INPROGRESS", 1012: "FOTA_NO_MEMORY", 1013: "FOTA_FILE_SIZE_TOO_LARGE", 1014: "FOTA_PARAM_SIZE_INVALID", } class OTA(object): def __init__(self, module, file_info): self.module = module self.file_info = file_info self.fota_queue = Queue(maxsize=4) def start(self): if self.module == SYSNAME: return self.start_fota() elif self.module == PROJECT_NAME: return self.start_sota() else: log.error('OTA Module %s Is Not Error!' % self.module) return False def start_fota(self): fota_obj = fota() url1 = self.file_info[0]['url'] url2 = self.file_info[1]['url'] if len(self.file_info) > 1 else '' res = fota_obj.httpDownload(url1=url1, url2=url2, callback=self.fota_cb) if res == 0: fota_res = self.fota_queue.get() return fota_res else: return False def fota_cb(self, args): down_status = args[0] down_process = args[1] if down_status in (0, 1): # TODO: Report To Cloud Upgrade Process. log.debug('DownStatus: %s [%s][%s%%]' % (down_status, '=' * down_process, down_process)) elif down_status == 2: # Download Over & Check Over, To Power Restart Update. self.fota_queue.put(True) else: log.error('Down Failed. Error Code [%s] %s' % (down_process, FOTA_ERROR_CODE.get(down_process, down_process))) self.fota_queue.put(False) def start_sota(self): ota_module_obj = SotaDownloadUpgrade() for file in self.file_info: if ota_module_obj.app_fota_down(file['url']): if ota_module_obj.file_update(file['md5']): continue else: return False else: return False ota_module_obj.sota_set_flag() return True class SotaDownloadUpgrade(object): def __init__(self, parent_dir="/usr/.updater/usr/"): self.fp_file = "/usr/sotaFile.tar.gz" self.file_list = [] self.parent_dir = parent_dir self.unzipFp = 0 self.hash_obj = uhashlib.md5() def write_update_data(self, data): with open(self.fp_file, "wb+") as fp: fp.write(data) self.hash_obj.update(data) def app_fota_down(self, url): app_fota_obj = app_fota.new() res = app_fota_obj.download(url, self.fp_file) if res == 0: uos.rename('/usr/.updater' + self.fp_file, self.fp_file) self.hash_obj = uhashlib.md5() with open(self.fp_file, "rb+") as fp: for fpi in fp.readlines(): self.hash_obj.update(fpi) return True else: return False def __get_file_size(self, data): size = data.decode('ascii') size = size.rstrip('\0') if (len(size) == 0): return 0 size = int(size, 8) return size def __get_file_name(self, name): fileName = name.decode('ascii') fileName = fileName.rstrip('\0') return fileName def file_update(self, md5_value): md5Data = ubinascii.hexlify(self.hash_obj.digest()) md5Data = md5Data.decode('ascii') md5Value = eval(md5_value) if not isinstance(md5_value, str) else md5_value log.debug("DMP Calc MD5 Value: %s, Device Calc MD5 Value: %s" % (md5Value, md5Data)) if (md5Value != md5Data): log.error("MD5 Verification Failed") return log.debug("MD5 Verification Success.") ota_file = open(self.fp_file, "wb+") ota_file.seek(10) self.unzipFp = uzlib.DecompIO(ota_file, -15) log.debug('Unzip File Success.') ql_fs.mkdirs(self.parent_dir) try: while True: data = self.unzipFp.read(0x200) if not data: log.debug("Read File Size Zore.") break size = self.__get_file_size(data[124:135]) fileName = self.__get_file_name(data[:100]) log.debug("File Name: %s, File Size: %s" % (fileName, size)) if not size: if len(fileName): log.debug("Create File Dir: %s" % self.parent_dir + fileName) ql_fs.mkdirs(self.parent_dir + fileName) else: log.debug("Have No File Unzip.") break else: log.debug("File %s Write Size %s" % (self.parent_dir + fileName, size)) fp = open(self.parent_dir + fileName, "wb+") fileSize = size while fileSize: data = self.unzipFp.read(0x200) if (fileSize < 0x200): fp.write(data[:fileSize]) fileSize = 0 fp.close() self.file_list.append({"fileName": "/usr/" + fileName, "size": size}) break else: fileSize -= 0x200 fp.write(data) for fileName in self.file_list: app_fota_download.update_download_stat("/usr/.updater" + fileName["fileName"], fileName["fileName"], fileName["size"]) ota_file.close() log.debug("Remove %s" % self.fp_file) uos.remove(self.fp_file) except Exception as e: log.error("Unpack Error: %s" % e) return False return True def sota_set_flag(self): app_fota_download.set_update_flag() class OTAFileClear(object): def __init__(self): self.usrList = uos.ilistdir("/usr/") def __remove_updater_dir(self, path): dirList = uos.ilistdir(path) for fileInfo in dirList: if fileInfo[1] == 0x4000: self.__remove_updater_dir("%s/%s" % (path, fileInfo[0])) else: log.debug("remove file name: %s/%s" % (path, fileInfo[0])) uos.remove("%s/%s" % (path, fileInfo[0])) log.debug("remove dir name: %s" % path) uos.remove(path) def file_clear(self): for fileInfo in self.usrList: if fileInfo[0] == ".updater": self.__remove_updater_dir("/usr/.updater") elif fileInfo[0] == "sotaFile.tar.gz": log.debug("remove update file sotaFile.tar.gz") uos.remove("/usr/sotaFile.tar.gz")