CameraMachine/python/custom_plugins/plugins/micropython_update/ampy_plugin.py

import json
import time

if __name__ == '__main__':
    raise "不允许在当前目录下运行"

import os
import settings
from ampy.pyboard import Pyboard
import ampy.files
from module.view_control.main_window.data_main import DataMode
from module.base_mode.base import calculate_sha256, download_file, compare_two_times, get_modified_time
import concurrent.futures

plugins_name = "AMPY插件"


class AmpyScribe(object):
    def __init__(self, port_name="COM37", baud_rate=115200, windows=None):
        self.port_name = port_name
        self.baud_rate = baud_rate
        self.windows = windows
        self.pyb = Pyboard(self.port_name, baudrate=self.baud_rate)

    def send_log(self, text):
        if self.windows:
            self.windows.send_log(text)

    # 运行本地脚本
    def run_remote_script(self, script_path):
        """Run a script on the remote device."""
        try:
            self.pyb.enter_raw_repl()
            result = self.pyb.execfile(script_path)
            print(result.decode())
            return result.decode()
        except Exception as e:
            print(f"An error occurred while running the script: {e}")
            return None

    def close(self):
        if self.pyb is not None:
            self.pyb.close()
            self.pyb = None


    # 上传文件
    def upload_file(self, local_path, remote_path, local_path_is_file=True):
        """
        local_path:电脑本地路径
        remote_path:单片机路径，/开头表示根目录
        """
        try:
            if local_path_is_file:
                # 打开文件并读取内容
                with open(local_path, 'rb') as file:
                    content = file.read()
            else:
                content = local_path

            # 创建一个 Files 对象用于文件操作
            files = ampy.files.Files(self.pyb)
            # 将文件内容写入 MicroPython 设备
            files.put(remote_path, content)
        except Exception as e:
            print(f"An error occurred while upload_file: {e}")

    # 获取远程文件
    def get_file_from_device(self, remote_filename, local_filename):
        try:
            # 创建一个 Files 对象用于文件操作
            files = ampy.files.Files(self.pyb)
            # 获取远程文件内容
            content = files.get(remote_filename)
            # 将内容写入本地文件
            with open(local_filename, 'wb') as file:
                file.write(content)
            print(f"文件 {remote_filename} 下载成功并保存为 {local_filename}")
        except Exception as e:
            print(f"get_file_from_device 发生错误: {e}")


# 远程更新脚本
class RemoteUpdate(object):
    # instance = None
    # init_flag = None

    def __init__(self, port_name=None, top_windows=None, windows=None, is_test=False):
        # if self.init_flag:
        #     return
        # else:
        #     self.init_flag = True
        self.is_test = is_test
        self.port_name = port_name
        self.top_windows = top_windows
        self.windows = windows
        self.ampy_scribe = None
        self.root = os.getcwd()
        self.value = 0
        # 基础数据mode
        self.data_mode = DataMode()
        self.mcu_code_path = "{}\micropython_online_code\mcu_code".format(os.getcwd())

    def send_log(self, text):
        print(text)
        pass

    def show_text_info(self, text):
        print(text)
        if self.windows:
            self.windows.text_info_sign.emit({"text": text})

    def show_progress_bar(self, add_value=None, value=None):
        if self.windows:
            if value:
                self.windows.progress_bar_sign.emit(value)
            else:
                self.value += add_value
                self.windows.progress_bar_sign.emit(self.value)

    def check_windows_state(self):
        if self.windows:
            if self.windows.state != 1:
                self.show_text_info("已取消")
                raise "主动停止"

    def run(self):
        """
        步骤：
        1、进行设备终止,并连接
        2、获取远程文件的md5合集
        3、比对线上合集
        4、整理差异内容，以及对应文件的has校验码
        5、下发文件到临时目录
        6、运行远程更新脚本
        7、自动重启
        """
        self.show_progress_bar(add_value=5)
        self.show_text_info("开始检查处理，停止MCU")
        self.stop_mcu()
        time.sleep(0.5)
        self.connect_pyboard()
        if self.ampy_scribe is None:
            self.show_text_info("串口链接失败，已退出")
            return

        # 删除远程的临时文件
        self.show_text_info("开始 删除远程的临时文件")
        self.remove_remote_files()
        self.show_progress_bar(add_value=5)
        self.show_text_info("开始 获取MCU文件的md5合集")
        # 获取远程文件的md5合集
        micropython_file_dict = self.get_remote_has()
        if micropython_file_dict is None:
            self.show_text_info("获取MCU文件的md5失败")
            return

        print("==============micropython_file_dict")
        print(micropython_file_dict)

        # 获取线上的数据
        self.show_text_info("开始 获取线上的数据")
        online_files_dict = self.check_resources_download()
        self.show_progress_bar(add_value=5)
        if not online_files_dict:
            self.show_text_info("开始 获取线上的数据  失败")
            return
        # 线上数据与单片机文件进行比对
        print("==============online_files_dict")
        print(online_files_dict)

        # 生成差异文件
        self.show_text_info("比对差异文件数据")
        differential_files_dict = self.get_differential_files(micropython_file_dict=micropython_file_dict,
                                                              online_files_dict=online_files_dict)
        print("==============differential_files_dict")
        print(differential_files_dict)
        ## 动态创建文件夹路径
        # self.create_all_folder()
        if not differential_files_dict:
            self.show_progress_bar(value=95)
            self.show_text_info("MCU文件为最新无需更新，开始重启")
            self.restart()
            self.show_text_info("开始断开连接")
            self.ampy_scribe.close()
            # 尝试连接MCU
            self.mcu_connect()
            self.show_text_info("完成")
            self.show_progress_bar(value=100)
            return
        # 移动差异文件
        self.show_text_info("开始下发差异文件，总数：{}".format(len(differential_files_dict)))
        to_move_files = self.upload_all_file(differential_files_dict)
        if not to_move_files:
            self.show_text_info("下发差异文件失败")
            return

        # 制作更新脚本
        self.show_text_info("开始MCU主程序更新")
        result = self.update_from_temp()
        if result:
            if "更新成功" in result:
                self.show_text_info("更新成功，即将重启")
                print("单片机程序为最新,5秒后重启")
                # 重启
                self.restart()
                self.show_text_info("开始断开连接")
                self.ampy_scribe.close()
                # 尝试连接MCU
                self.mcu_connect()
                self.show_progress_bar(value=100)
                return

    # 制作更新脚本
    def update_from_temp(self):
        print("制作更新脚本")
        path = r"{}\custom_plugins\plugins\micropython_update\other_micropython_code\update_from_temp.py".format(
            os.getcwd())
        return self.ampy_scribe.run_remote_script(path)

    # 创建所有文件夹路径
    def create_all_folder(self):
        path = r"{}\custom_plugins\plugins\micropython_update\other_micropython_code\create_all_folders.py".format(
            os.getcwd())
        self.ampy_scribe.run_remote_script(path)
        print("创建文件夹路径已完成")

    # 批量移动文件到单片机
    def upload_all_file(self, files_dict):
        # k 示例：sensor/other_sensor/nrf24.py    settings.py
        to_move_files = {}
        f = True
        total_s = 100 - self.value - 5
        total_len = len(files_dict)
        step = int(total_s / total_len)

        for relative_file_path, value in files_dict.items():
            local_path = "{}\{}".format(self.mcu_code_path, relative_file_path)
            # 注意路径格式 / 开头
            remote_path = r"/temp/{}".format(relative_file_path)
            print("移动文件", local_path, remote_path)
            try:
                self.ampy_scribe.upload_file(local_path, remote_path)
                # to_move_files[remote_path] = value["file_sha256"]
                # 使用本地has
                to_move_files[remote_path] = calculate_sha256(local_path)
                self.show_text_info("文件：{}下发成功".format(relative_file_path))
            except BaseException as e:
                self.show_text_info("文件：{}下发失败：{}".format(relative_file_path, e))
                f = False
                return None
            self.show_progress_bar(add_value=step)

        if f:
            # 创建配置文件
            print("创建配置文件")
            self.show_text_info("创建配置md5文件 并开始下发")
            _to_move_files_config = json.dumps(to_move_files)
            self.ampy_scribe.upload_file(local_path=_to_move_files_config,
                                         remote_path=r"/temp/_to_move_files_config.json", local_path_is_file=False)
            self.show_text_info("创建配置md5文件 并下发成功")
        if f:
            return to_move_files
        else:
            return None

    # 生成差异文件
    def get_differential_files(self, micropython_file_dict: dict, online_files_dict: dict):
        # 生成差异文件
        ignore_path = ["machine.py", "boot.py"]  # "main.py"
        differential_files_dict = {}
        for online_file_path, value in online_files_dict.items():
            if online_file_path in ignore_path:
                continue
            if online_file_path not in micropython_file_dict:
                differential_files_dict[online_file_path] = {"file_sha256": value["file_sha256"]}
                print("文件不存在：{}".format(online_file_path))
            else:
                if value["file_sha256"] != micropython_file_dict[online_file_path]["file_sha256"]:
                    differential_files_dict[online_file_path] = {"file_sha256": value["file_sha256"]}
                else:
                    print("文件相同：{}".format(online_file_path))

        return differential_files_dict

    # 删除远程的临时文件
    def remove_remote_files(self):
        script_path = r"{}\custom_plugins\plugins\micropython_update\other_micropython_code\delete_all_temp.py".format(
            self.root)
        result = self.ampy_scribe.run_remote_script(script_path)
        print(result)

    def run_run_remote_script_restart(self):
        script_path = r"{}\custom_plugins\plugins\micropython_update\other_micropython_code\restart.py".format(
            self.root)
        self.ampy_scribe.run_remote_script(script_path)

    # 重启单片机
    def restart(self):
        try:
            # 使用上下文管理器确保线程池会正确关闭
            with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
                future = executor.submit(self.run_run_remote_script_restart)
                try:
                    result = future.result(timeout=1)  # 等待最多5秒
                except BaseException as e:
                    print("自动超时")
                finally:
                    # 如果需要立即释放资源，可以显式地调用 shutdown。
                    # 在 with 语句块结束时也会自动调用 shutdown。
                    # executor.shutdown(wait=False)
                    print("shutdown")
            print("-----------end--------------")
        except BaseException as e:
            print(e)
            print("程序遇到了一个致命错误")

    def mcu_connect(self):
        self.show_text_info("5秒后 尝试连接MCU进行恢复")
        time.sleep(5)
        self.top_windows.mcu.to_connect_com(port_name=self.port_name)
        self.show_progress_bar(value=100)

    # 检查file_sha256 并进行下载更新文件
    def check_resources_download(self):
        if self.is_test is False:
            if not settings.IsLogin:
                print("当前未登录")
                return
        # 请求获取配置文件
        online_data = self.data_mode.get_all_resource_config(plugins_name="plugins_micropython")
        # print("check_resources_download")
        # print(data)
        if not online_data:
            text = "check_resources_download  data为空"
            self.send_log(text)
        elif "data" not in online_data:
            text = "check_resources_download  data不存在"
            self.send_log(text)
        else:
            # 步骤：
            # 1、获取线上数据
            # 2、如本地MD5不同，但是修改时间比线上的最新，则也不更新下载。
            for relative_file_path, value in online_data["data"].items():
                file_path = "{}\{}".format(self.mcu_code_path, relative_file_path)
                if os.path.exists(file_path):
                    file_has256 = calculate_sha256(file_path)
                    if file_has256 != value["file_sha256"]:
                        file_modified_time = get_modified_time(file_path)
                        if compare_two_times(file_modified_time, online_data["update_time"]) == "left_new":
                            continue
                        else:
                            text = "has256 不同 开始下载：{}".format(file_path)
                            self.send_log(text)
                            download_file(url=value["url"], file_path=file_path)
                    else:
                        pass
                else:
                    text = "文件不存在 开始下载：{}".format(file_path)
                    self.send_log(text)
                    download_file(url=value["url"], file_path=file_path)
        if online_data:
            return online_data["data"]
        return None

    def get_remote_has(self):
        self.ampy_scribe: AmpyScribe
        script_path = r"{}\custom_plugins\plugins\micropython_update\other_micropython_code\get_remote_has256.py".format(
            self.root)
        result = self.ampy_scribe.run_remote_script(script_path)
        if result is None:
            return None
        has_dict = {}
        try:
            result_list = result.split("\n")
            for i in result_list:
                if "@" not in i:
                    continue
                path, _has = i.split("@")
                has_dict[path[1:]] = {"file_sha256": _has.replace("\r", "")}
            return has_dict
        except BaseException as e:
            return None

    def stop_mcu(self):
        if self.is_test:
            return
        if self.ampy_scribe is None:
            self.top_windows.mcu.stop_mcu()

    def connect_pyboard(self):
        if self.is_test:
            self.ampy_scribe = AmpyScribe(port_name=self.port_name, windows=None)
            print("ampy_scribe 接连成功")
            return

        if self.ampy_scribe is None:
            if self.port_name:
                pass
            else:
                self.port_name = self.top_windows.mcu.port_name

            print("port_name:", self.port_name)
            if self.port_name:
                self.top_windows.mcu.close_connect()
                time.sleep(3)
                try:
                    self.ampy_scribe = AmpyScribe(port_name=self.port_name, windows=None)
                    print("ampy_scribe 接连成功")
                except BaseException as e:
                    print("ampy_scribe 连接失败", e)
                    self.ampy_scribe = None

    def __new1__(cls, *args, **kwargs):
        """如果当前没有实例时，调用父类__new__方法，生成示例，有则返回保存的内存地址。"""
        if not cls.instance:
            cls.instance = super().__new__(cls)
        return cls.instance