demo.tracker-v2/code/ota.py

# 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__)


def 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 != -1:
            log.debug('DownStatus: %s [%s][%s%%]' % (down_status, '=' * down_process, down_process))
            if down_status == 0 and down_process == 100:
                self.fota_queue.put(True)
                # TODO: Report To Cloud Upgrade Process.
        else:
            log.error('Down Failed. Error Code: %s' % 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()
        fp = open(self.fp_file, "wb+")
        fp.close()
        res = app_fota_obj.download(url, self.fp_file)
        if res == 0:
            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)
        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 /usr/sotaFile.tar.gz")
            uos.remove("/usr/sotaFile.tar.gz")
        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")