2317 lines
93 KiB
Python
2317 lines
93 KiB
Python
from __future__ import annotations
|
||
|
||
import getpass
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import platform
|
||
import re
|
||
import sys
|
||
import threading
|
||
import tempfile
|
||
import time
|
||
import tkinter as tk
|
||
import uuid
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from tkinter import messagebox, scrolledtext, ttk
|
||
from typing import Any, Callable, Iterable, Optional, Tuple
|
||
|
||
from factory_common.firmware_envelope import (
|
||
FirmwareEnvelopeError,
|
||
decrypt_firmware_blob,
|
||
load_firmware_decrypt_key,
|
||
)
|
||
from factory_common.signing import SignatureError, verify_file_signature
|
||
from factory_private_config import (
|
||
EncryptedConfigError,
|
||
load_private_config_from_path,
|
||
resolve_private_config_path,
|
||
)
|
||
|
||
from .config import (
|
||
_load_flash_private_config,
|
||
_ui_text,
|
||
)
|
||
from .constants import (
|
||
DEFAULT_BAUD,
|
||
DEFAULT_FLASH_MAX_ATTEMPTS,
|
||
DEFAULT_FLASH_SECURE_MAX_ATTEMPTS,
|
||
DEFAULT_READY_TIMEOUT_S,
|
||
)
|
||
from .flash_utils import (
|
||
_extract_json_blob,
|
||
_parse_offset,
|
||
_sha256_hex,
|
||
_value_to_int,
|
||
_wipe_bytearray,
|
||
build_flash_entries,
|
||
collect_error_hints,
|
||
detect_boot_chain_header_state,
|
||
detect_secure_boot_block,
|
||
extract_mac_from_output,
|
||
evaluate_post_flash_status,
|
||
get_efuse_flag,
|
||
recommend_mode_from_json,
|
||
)
|
||
from .models import (
|
||
DeviceCheckConfig,
|
||
FlashEntryReport,
|
||
FlashPrivateConfig,
|
||
FlashReport,
|
||
FlashSource,
|
||
)
|
||
from .path_utils import _display_path, get_base_dir, is_frozen, resolve_python, select_bin_dir
|
||
from .redact import _redact_sensitive, _sanitize_filename
|
||
from .serial_utils import ensure_pyserial, list_serial_ports, pick_preferred_port
|
||
from .time_utils import _now_iso
|
||
from .tooling import run_command, run_esptool_api, run_tool_module
|
||
|
||
|
||
class FlashGUI:
|
||
def __init__(self) -> None:
|
||
self.base_dir = get_base_dir()
|
||
self.frozen = is_frozen()
|
||
self.python_cmd: Optional[Path] = None if self.frozen else resolve_python(self.base_dir)
|
||
self.run_record: Optional[dict] = None
|
||
self.current_device_id: Optional[str] = None
|
||
self.device_id_hint: Optional[str] = None
|
||
self._post_flash_reset_done = False
|
||
|
||
self.ready_timeout_s = DEFAULT_READY_TIMEOUT_S
|
||
self.flash_max_attempts = DEFAULT_FLASH_MAX_ATTEMPTS
|
||
self.flash_secure_max_attempts = DEFAULT_FLASH_SECURE_MAX_ATTEMPTS
|
||
|
||
self.private_cfg_path = resolve_private_config_path(self.base_dir)
|
||
if self.private_cfg_path is None or not self.private_cfg_path.exists():
|
||
raise FileNotFoundError(
|
||
"缺少工厂私有配置文件（factory_private.json）。请设置环境变量 "
|
||
"FACTORY_PRIVATE_CONFIG/FACTORY_PRIVATE_CONFIG_PATH 指定配置路径。"
|
||
)
|
||
self.private_cfg_loaded = False
|
||
|
||
self.ui_title = "产线烧录工具"
|
||
self.ui_subtitle = "插入设备后点击开始，系统将自动完成流程"
|
||
self.label_port = "串口"
|
||
self.label_secure = "启动信息"
|
||
self.label_flash_enc = "写入信息"
|
||
self.label_key = "配置信息"
|
||
self.label_recommend = "下一步"
|
||
self.label_log = "日志"
|
||
self.plain_text = "模式 A"
|
||
self.app_text = "模式 C"
|
||
self.safe_log = True
|
||
|
||
raw_cfg: Optional[dict] = None
|
||
try:
|
||
raw_cfg = load_private_config_from_path(self.private_cfg_path)
|
||
except EncryptedConfigError:
|
||
raw_cfg = None
|
||
if raw_cfg is not None:
|
||
self._apply_private_config(raw_cfg)
|
||
else:
|
||
self.private_cfg = self._build_placeholder_config()
|
||
|
||
self.root = tk.Tk()
|
||
self._apply_window_icon()
|
||
ui_cfg = self.private_cfg.ui
|
||
self.ui_title = _ui_text(ui_cfg, "flash_title", "产线烧录工具")
|
||
self.ui_subtitle = _ui_text(ui_cfg, "flash_subtitle", "插入设备后点击开始，系统将自动完成流程")
|
||
self.label_port = _ui_text(ui_cfg, "label_port", "串口")
|
||
self.label_secure = _ui_text(ui_cfg, "label_secure", "启动信息")
|
||
self.label_flash_enc = _ui_text(ui_cfg, "label_flash_enc", "写入信息")
|
||
self.label_key = _ui_text(ui_cfg, "label_key", "配置信息")
|
||
self.label_recommend = _ui_text(ui_cfg, "label_recommend", "下一步")
|
||
self.label_log = _ui_text(ui_cfg, "label_log", "日志")
|
||
safe_log_value = ui_cfg.get("safe_log")
|
||
self.safe_log = bool(safe_log_value) if isinstance(safe_log_value, bool) else True
|
||
|
||
self.root.title(self.ui_title)
|
||
self.root.geometry("820x620")
|
||
self.root.minsize(760, 560)
|
||
self.root.resizable(True, True)
|
||
|
||
self.plain_text = _ui_text(ui_cfg, "flash_mode_plain", "模式 A")
|
||
self.app_text = _ui_text(ui_cfg, "flash_mode_app", "模式 C")
|
||
|
||
self.port_var = tk.StringVar(value="COM3")
|
||
self.status_var = tk.StringVar(value="就绪。")
|
||
self.recommend_var = tk.StringVar(value="插入设备后点击开始。")
|
||
self.hint_var = tk.StringVar(value="")
|
||
self.mode_status_var = tk.StringVar(value="")
|
||
self.key_status_var = tk.StringVar(value="")
|
||
self.progress_value = tk.DoubleVar(value=0.0)
|
||
self.active_mode_label = ""
|
||
self.suppress_ready_status = False
|
||
self.checked_port: Optional[str] = None
|
||
self._last_port_value = ""
|
||
|
||
self.worker: Optional[threading.Thread] = None
|
||
self.is_busy = False
|
||
self.summary_attempted = False
|
||
self.summary_ok = False
|
||
self.last_rec_mode: Optional[str] = None
|
||
self.last_flash_app_only: Optional[bool] = None
|
||
self.secure_boot_blocked = False
|
||
self.secure_boot_enabled: Optional[bool] = None
|
||
self.boot_chain_present: Optional[bool] = None
|
||
self.boot_chain_probe_reason: Optional[str] = None
|
||
self.download_mode_disabled: Optional[bool] = None
|
||
self.secure_download_enabled: Optional[bool] = None
|
||
self.manual_encrypt_disabled: Optional[bool] = None
|
||
self.last_flash_block_reason: Optional[str] = None
|
||
self.card_port_var = tk.StringVar(value="未选择")
|
||
self.card_secure_boot_var = tk.StringVar(value="未检测")
|
||
self.card_flash_enc_var = tk.StringVar(value="未检测")
|
||
self.card_recommend_var = tk.StringVar(value="待检测")
|
||
self._build_ui()
|
||
self.preflight_check()
|
||
self._auto_set_port(initial=True)
|
||
self._last_port_value = self._normalize_port(self.port_var.get())
|
||
self.port_var.trace_add("write", self._on_port_change)
|
||
|
||
def _apply_window_icon(self) -> None:
|
||
icon_path = self._resolve_icon_path()
|
||
if not icon_path.exists():
|
||
return
|
||
try:
|
||
self.root.iconbitmap(default=str(icon_path))
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
def _resolve_icon_path(self) -> Path:
|
||
if self.frozen and hasattr(sys, "_MEIPASS"):
|
||
return Path(getattr(sys, "_MEIPASS")) / "talkingq_logo_256x256.ico"
|
||
return self.base_dir / "talkingq_logo_256x256.ico"
|
||
|
||
def _build_placeholder_config(self) -> FlashPrivateConfig:
|
||
device_check = DeviceCheckConfig(
|
||
crypt_cnt_fields=("SPI_BOOT_CRYPT_CNT",),
|
||
secure_boot_field="SECURE_BOOT_EN",
|
||
secure_boot_expected=None,
|
||
crypt_cnt_expect=None,
|
||
flash_key_purpose_match=None,
|
||
secure_boot_key_purpose_match=None,
|
||
key_read_disable_field="RD_DIS",
|
||
)
|
||
flash_cfg = FlashPrivateConfig(
|
||
chip=os.getenv("FACTORY_FLASH_CHIP", "esp32s3"),
|
||
baud=str(DEFAULT_BAUD),
|
||
bin_dir=Path("bin"),
|
||
layout=[],
|
||
app_bin_name="",
|
||
app_partition_offset="0x0",
|
||
flash_args={
|
||
"flash_mode": "dio",
|
||
"flash_freq": "80m",
|
||
"flash_size": "keep",
|
||
"before": "default-reset",
|
||
"after": "no-reset",
|
||
},
|
||
device_check=device_check,
|
||
ui={},
|
||
)
|
||
return flash_cfg
|
||
|
||
def _apply_private_config(self, raw_cfg: dict) -> None:
|
||
self.private_cfg = _load_flash_private_config(self.base_dir, raw_cfg)
|
||
self.private_cfg_loaded = True
|
||
|
||
ui_cfg = self.private_cfg.ui if isinstance(self.private_cfg.ui, dict) else {}
|
||
self.ui_title = _ui_text(ui_cfg, "flash_title", self.ui_title)
|
||
self.ui_subtitle = _ui_text(ui_cfg, "flash_subtitle", self.ui_subtitle)
|
||
self.label_port = _ui_text(ui_cfg, "label_port", self.label_port)
|
||
self.label_secure = _ui_text(ui_cfg, "label_secure", self.label_secure)
|
||
self.label_flash_enc = _ui_text(ui_cfg, "label_flash_enc", self.label_flash_enc)
|
||
self.label_key = _ui_text(ui_cfg, "label_key", self.label_key)
|
||
self.label_recommend = _ui_text(ui_cfg, "label_recommend", self.label_recommend)
|
||
self.label_log = _ui_text(ui_cfg, "label_log", self.label_log)
|
||
safe_log_value = ui_cfg.get("safe_log")
|
||
if isinstance(safe_log_value, bool):
|
||
self.safe_log = safe_log_value
|
||
self.plain_text = _ui_text(ui_cfg, "flash_mode_plain", self.plain_text)
|
||
self.app_text = _ui_text(ui_cfg, "flash_mode_app", self.app_text)
|
||
|
||
if self.run_record and not self.run_record.get("firmware_version"):
|
||
self.run_record["firmware_version"] = self.private_cfg.app_bin_name
|
||
|
||
def _ensure_private_config(self, port: str, show_dialog: bool) -> bool:
|
||
if self.private_cfg_loaded:
|
||
return True
|
||
if self.private_cfg_path is None or not self.private_cfg_path.exists():
|
||
if show_dialog:
|
||
self._notify("error", "缺少配置", "未找到私有配置文件。")
|
||
return False
|
||
try:
|
||
raw_cfg = load_private_config_from_path(self.private_cfg_path)
|
||
except EncryptedConfigError:
|
||
self.log("[错误] 检测到加密私有配置。当前仅支持明文配置文件。")
|
||
if show_dialog:
|
||
self._notify(
|
||
"error",
|
||
"不支持加密配置",
|
||
"当前仅支持明文 conf/factory_private.json。",
|
||
)
|
||
return False
|
||
except (PermissionError, ValueError, FileNotFoundError) as exc:
|
||
self.log(f"[错误] 私有配置加载失败：{exc}")
|
||
if show_dialog:
|
||
self._notify("error", "配置加载失败", str(exc))
|
||
return False
|
||
|
||
self._apply_private_config(raw_cfg)
|
||
self._refresh_key_status()
|
||
if hasattr(self, "root"):
|
||
self.root.after(0, lambda: self.root.title(self.ui_title))
|
||
return True
|
||
|
||
def _refresh_mode_status(self) -> None:
|
||
status = "运行模式：本地离线"
|
||
self.root.after(0, lambda: self.mode_status_var.set(status))
|
||
|
||
def _read_chip_mac(self, port: str) -> Optional[str]:
|
||
if not port:
|
||
return None
|
||
candidates: list[Optional[str]] = []
|
||
if self.private_cfg_loaded:
|
||
candidates.append(self.private_cfg.chip)
|
||
candidates.extend(["auto", None])
|
||
tried: set[Optional[str]] = set()
|
||
for chip in candidates:
|
||
if chip in tried:
|
||
continue
|
||
tried.add(chip)
|
||
if chip:
|
||
args = ["--chip", chip, "--port", port, "read_mac"]
|
||
else:
|
||
args = ["--port", port, "read_mac"]
|
||
code, output = self.run_tool("esptool", args, stream=False)
|
||
if code != 0:
|
||
continue
|
||
mac = extract_mac_from_output(output)
|
||
if mac:
|
||
return mac
|
||
return None
|
||
|
||
def _refresh_key_status(self) -> None:
|
||
status = "配置签名：已启用"
|
||
self.root.after(0, lambda: self.key_status_var.set(status))
|
||
|
||
def _build_ui(self) -> None:
|
||
self.style = ttk.Style(self.root)
|
||
for theme in ("clam", "vista"):
|
||
if theme in self.style.theme_names():
|
||
self.style.theme_use(theme)
|
||
break
|
||
self.root.option_add("*Font", ("Segoe UI", 10))
|
||
palette = {
|
||
"bg": "#eef4fb",
|
||
"panel": "#f8fbff",
|
||
"text": "#15233b",
|
||
"muted": "#5a6b85",
|
||
"accent": "#1a73e8",
|
||
"accent_soft": "#d6e7ff",
|
||
"border": "#c9d7ee",
|
||
}
|
||
self.root.configure(bg=palette["bg"])
|
||
self.style.configure(".", background=palette["bg"], foreground=palette["text"])
|
||
self.style.configure("TFrame", background=palette["bg"])
|
||
self.style.configure("Root.TFrame", background=palette["bg"])
|
||
self.style.configure("Panel.TFrame", background=palette["panel"])
|
||
self.style.configure("TLabelframe", background=palette["panel"], bordercolor=palette["border"])
|
||
self.style.configure("TLabelframe.Label", background=palette["panel"], foreground=palette["text"])
|
||
self.style.configure("TLabel", background=palette["bg"], foreground=palette["text"])
|
||
self.style.configure("Root.TLabel", background=palette["bg"], foreground=palette["text"])
|
||
self.style.configure("Panel.TLabel", background=palette["panel"], foreground=palette["text"])
|
||
self.style.configure(
|
||
"Header.TLabel",
|
||
font=("Segoe UI", 14, "bold"),
|
||
foreground=palette["accent"],
|
||
background=palette["bg"],
|
||
)
|
||
self.style.configure(
|
||
"SubHeader.TLabel",
|
||
font=("Segoe UI", 9),
|
||
foreground=palette["muted"],
|
||
background=palette["bg"],
|
||
)
|
||
self.style.configure(
|
||
"Step.TLabel",
|
||
font=("Segoe UI", 9),
|
||
foreground=palette["muted"],
|
||
background=palette["bg"],
|
||
)
|
||
self.style.configure(
|
||
"StepActive.TLabel",
|
||
font=("Segoe UI", 9, "bold"),
|
||
foreground=palette["accent"],
|
||
background=palette["bg"],
|
||
)
|
||
self.style.configure(
|
||
"Action.TButton",
|
||
padding=(14, 6),
|
||
font=("Segoe UI", 10, "bold"),
|
||
foreground=palette["text"],
|
||
background=palette["accent_soft"],
|
||
)
|
||
self.style.map(
|
||
"Action.TButton",
|
||
background=[("active", "#bfd7ff"), ("pressed", "#a8c8ff")],
|
||
)
|
||
self.style.configure("TButton", padding=(10, 4), background=palette["panel"])
|
||
self.style.map("TButton", background=[("active", "#e6f0ff")])
|
||
self.style.configure("Disabled.TButton", foreground="#777777")
|
||
self.style.configure("Section.TLabelframe.Label", font=("Segoe UI", 10, "bold"))
|
||
self.style.configure("TEntry", fieldbackground=palette["panel"])
|
||
|
||
frame = ttk.Frame(self.root, padding=14, style="Root.TFrame")
|
||
frame.pack(fill="both", expand=True)
|
||
|
||
header = ttk.Frame(frame, style="Root.TFrame")
|
||
header.pack(fill="x", pady=(0, 8))
|
||
ttk.Label(
|
||
header,
|
||
text=self.ui_title,
|
||
style="Header.TLabel",
|
||
).pack(anchor="w")
|
||
ttk.Label(
|
||
header,
|
||
text=self.ui_subtitle,
|
||
style="SubHeader.TLabel",
|
||
).pack(anchor="w", pady=(2, 0))
|
||
|
||
port_section = ttk.LabelFrame(frame, text="1. 串口连接", padding=10, style="Section.TLabelframe")
|
||
port_section.pack(fill="x", pady=(0, 8))
|
||
port_row = ttk.Frame(port_section, style="Panel.TFrame")
|
||
port_row.pack(fill="x")
|
||
ttk.Label(port_row, text="串口：", style="Panel.TLabel").pack(side="left")
|
||
self.port_entry = ttk.Entry(port_row, textvariable=self.port_var, width=18)
|
||
self.port_entry.pack(side="left", padx=(6, 12))
|
||
self.detect_btn = ttk.Button(port_row, text="刷新串口", command=self.on_auto_detect)
|
||
self.detect_btn.pack(side="left", padx=(0, 12))
|
||
ttk.Label(
|
||
port_row,
|
||
text="提示：开始烧录时按住 BOOT，点一下 RESET。",
|
||
style="Panel.TLabel",
|
||
).pack(side="left", padx=(10, 0))
|
||
|
||
status_row = ttk.Frame(port_section, style="Panel.TFrame")
|
||
status_row.pack(fill="x", pady=(6, 0))
|
||
ttk.Label(status_row, textvariable=self.mode_status_var, style="Panel.TLabel").pack(
|
||
side="left"
|
||
)
|
||
ttk.Label(status_row, textvariable=self.key_status_var, style="Panel.TLabel").pack(
|
||
side="left", padx=(12, 0)
|
||
)
|
||
|
||
detect_section = ttk.LabelFrame(frame, text="2. 一键流程", padding=10, style="Section.TLabelframe")
|
||
detect_section.pack(fill="x", pady=(0, 8))
|
||
detect_row = ttk.Frame(detect_section, style="Panel.TFrame")
|
||
detect_row.pack(fill="x")
|
||
self.start_btn = ttk.Button(
|
||
detect_row, text="开始", command=self.on_start, style="Action.TButton"
|
||
)
|
||
self.start_btn.pack(side="left")
|
||
self.mode_label = ttk.Label(
|
||
detect_row,
|
||
textvariable=self.recommend_var,
|
||
foreground="#1a73e8",
|
||
font=("Segoe UI", 10, "bold"),
|
||
wraplength=520,
|
||
justify="left",
|
||
style="Panel.TLabel",
|
||
)
|
||
self.mode_label.pack(side="left", padx=(10, 0), fill="x", expand=True)
|
||
|
||
hint_frame = ttk.Frame(detect_section, style="Panel.TFrame")
|
||
hint_frame.pack(fill="x", pady=(6, 0))
|
||
ttk.Label(hint_frame, text="提示：", style="Panel.TLabel").pack(side="left", anchor="n")
|
||
self.hint_label = ttk.Label(
|
||
hint_frame,
|
||
textvariable=self.hint_var,
|
||
wraplength=680,
|
||
justify="left",
|
||
foreground="#5a6b85",
|
||
style="Panel.TLabel",
|
||
)
|
||
self.hint_label.pack(side="left", fill="x", expand=True)
|
||
|
||
log_section = ttk.LabelFrame(
|
||
frame, text=self.label_log, padding=10, style="Section.TLabelframe"
|
||
)
|
||
log_section.pack(fill="both", expand=True)
|
||
log_actions = ttk.Frame(log_section, style="Panel.TFrame")
|
||
log_actions.pack(fill="x", pady=(0, 6))
|
||
self.clear_btn = ttk.Button(log_actions, text="清空日志", command=self.clear_log)
|
||
self.clear_btn.pack(side="right")
|
||
self.copy_btn = ttk.Button(log_actions, text="复制日志", command=self.copy_log)
|
||
self.copy_btn.pack(side="right", padx=(0, 8))
|
||
|
||
self.log_box = scrolledtext.ScrolledText(
|
||
log_section,
|
||
height=16,
|
||
wrap="word",
|
||
background="#ffffff",
|
||
foreground="#1f2a33",
|
||
insertbackground="#1f2a33",
|
||
)
|
||
self.log_box.pack(fill="both", expand=True)
|
||
self.log_box.configure(state="disabled", font=("Consolas", 10))
|
||
|
||
status_frame = ttk.Frame(frame, style="Root.TFrame")
|
||
status_frame.pack(fill="x", pady=(8, 0))
|
||
ttk.Label(status_frame, text="提示：", style="Root.TLabel").pack(side="left")
|
||
self.status_label = ttk.Label(status_frame, textvariable=self.status_var, style="Root.TLabel")
|
||
self.status_label.pack(side="left", padx=(6, 0))
|
||
self.progress = ttk.Progressbar(
|
||
status_frame,
|
||
mode="indeterminate",
|
||
length=140,
|
||
maximum=100,
|
||
variable=self.progress_value,
|
||
)
|
||
self.progress.pack(side="right", padx=(8, 0))
|
||
self.progress.pack_forget()
|
||
self.update_flash_buttons()
|
||
|
||
def _notify(self, kind: str, title: str, message: str) -> None:
|
||
def _show() -> None:
|
||
if kind == "info":
|
||
messagebox.showinfo(title, message)
|
||
return
|
||
if kind == "warn":
|
||
messagebox.showwarning(title, message)
|
||
return
|
||
messagebox.showerror(title, message)
|
||
|
||
self.root.after(0, _show)
|
||
|
||
def log(self, message: str, timestamp: bool = True) -> None:
|
||
safe_message = _redact_sensitive(message)
|
||
|
||
def _append() -> None:
|
||
self.log_box.configure(state="normal")
|
||
prefix = ""
|
||
if timestamp and safe_message:
|
||
prefix = f"[{datetime.now().strftime('%H:%M:%S')}] "
|
||
line = prefix + safe_message if safe_message else ""
|
||
self.log_box.insert("end", line + "\n")
|
||
self.log_box.see("end")
|
||
self.log_box.configure(state="disabled")
|
||
|
||
self.root.after(0, _append)
|
||
|
||
def log_section(self, title: str) -> None:
|
||
self.log("", timestamp=False)
|
||
self.log(title, timestamp=False)
|
||
self.log("-" * len(title), timestamp=False)
|
||
|
||
def _resolve_operator(self) -> str:
|
||
for key in ("FACTORY_OPERATOR", "OPERATOR_ID", "OPERATOR", "USERNAME", "USER"):
|
||
value = os.getenv(key)
|
||
if isinstance(value, str) and value.strip():
|
||
return value.strip()
|
||
try:
|
||
return getpass.getuser()
|
||
except Exception:
|
||
return "unknown"
|
||
|
||
def _log_dir(self) -> Path:
|
||
env_path = os.getenv("FACTORY_LOG_DIR")
|
||
if isinstance(env_path, str) and env_path.strip():
|
||
return Path(env_path).expanduser()
|
||
if os.name == "nt":
|
||
base = os.getenv("LOCALAPPDATA") or os.getenv("APPDATA")
|
||
if base:
|
||
return Path(base) / "TalkingQFactory" / "logs"
|
||
return Path.home() / "AppData" / "Local" / "TalkingQFactory" / "logs"
|
||
if sys.platform == "darwin":
|
||
return Path.home() / "Library" / "Logs" / "TalkingQFactory"
|
||
return Path.home() / ".cache" / "talkingq-factory"
|
||
|
||
def _begin_run_record(self, port: str) -> None:
|
||
firmware_version = os.getenv("FACTORY_FIRMWARE_VERSION") or os.getenv("FIRMWARE_VERSION")
|
||
if isinstance(firmware_version, str) and firmware_version.strip():
|
||
firmware_version = firmware_version.strip()
|
||
else:
|
||
firmware_version = self.private_cfg.app_bin_name
|
||
run_id = uuid.uuid4().hex[:8]
|
||
self.run_record = {
|
||
"tool": "factory_flash_gui",
|
||
"run_id": run_id,
|
||
"started_at": _now_iso(),
|
||
"operator": self._resolve_operator(),
|
||
"host": platform.node(),
|
||
"port": port,
|
||
"device_id": None,
|
||
"firmware_version": firmware_version,
|
||
"steps": {},
|
||
}
|
||
|
||
def _record_error(self, stage: str, message: str, code: Optional[str] = None) -> None:
|
||
if not self.run_record:
|
||
return
|
||
self.run_record["error"] = {
|
||
"stage": stage,
|
||
"message": message,
|
||
"code": code,
|
||
}
|
||
|
||
def _record_step_result(
|
||
self,
|
||
step: str,
|
||
start_ts: float,
|
||
status: str,
|
||
details: Optional[dict] = None,
|
||
) -> None:
|
||
if not self.run_record:
|
||
return
|
||
started_at = datetime.fromtimestamp(start_ts, tz=timezone.utc).isoformat()
|
||
payload = {
|
||
"started_at": started_at,
|
||
"duration_s": int(max(0.0, time.time() - start_ts)),
|
||
"status": status,
|
||
}
|
||
if details:
|
||
payload.update(details)
|
||
self.run_record.setdefault("steps", {})[step] = payload
|
||
|
||
def _sanitize_run_record(self, record: dict) -> dict:
|
||
safe: dict = {
|
||
"tool": record.get("tool"),
|
||
"run_id": record.get("run_id"),
|
||
"device_id": record.get("device_id"),
|
||
"started_at": record.get("started_at"),
|
||
"ended_at": record.get("ended_at"),
|
||
"duration_s": record.get("duration_s"),
|
||
"success": record.get("success"),
|
||
}
|
||
error = record.get("error")
|
||
if isinstance(error, dict):
|
||
safe_error: dict = {}
|
||
for key in ("stage", "message", "code"):
|
||
value = error.get(key)
|
||
if isinstance(value, str):
|
||
safe_error[key] = _redact_sensitive(value)
|
||
elif value is not None:
|
||
safe_error[key] = value
|
||
if safe_error:
|
||
safe["error"] = safe_error
|
||
steps = {}
|
||
raw_steps = record.get("steps", {})
|
||
if isinstance(raw_steps, dict):
|
||
for step, payload in raw_steps.items():
|
||
if not isinstance(payload, dict):
|
||
continue
|
||
step_safe: dict = {}
|
||
for key in ("status", "duration_s", "reason", "attempts", "errors"):
|
||
if key not in payload:
|
||
continue
|
||
value = payload.get(key)
|
||
if isinstance(value, str):
|
||
step_safe[key] = _redact_sensitive(value)
|
||
elif isinstance(value, list):
|
||
step_safe[key] = [
|
||
_redact_sensitive(item) if isinstance(item, str) else item
|
||
for item in value
|
||
]
|
||
else:
|
||
step_safe[key] = value
|
||
if step_safe:
|
||
steps[step] = step_safe
|
||
if steps:
|
||
safe["steps"] = steps
|
||
return safe
|
||
|
||
def _write_run_record(self, record: dict) -> Optional[Path]:
|
||
log_dir = self._log_dir()
|
||
try:
|
||
log_dir.mkdir(parents=True, exist_ok=True)
|
||
except OSError as exc:
|
||
self.log(f"[警告] 无法创建记录目录: {exc}")
|
||
return None
|
||
record_id = record.get("device_id") or "unknown"
|
||
success = record.get("success")
|
||
if success is True:
|
||
status_tag = "ok"
|
||
elif success is False:
|
||
status_tag = "failed"
|
||
else:
|
||
status_tag = "unknown"
|
||
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
filename = f"{ts}_{_sanitize_filename(str(record_id))}_{status_tag}.json"
|
||
path = log_dir / filename
|
||
try:
|
||
path.write_text(json.dumps(record, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
except OSError as exc:
|
||
self.log(f"[警告] 记录落盘失败: {exc}")
|
||
return None
|
||
return path
|
||
|
||
def _finalize_run_record(self, success: bool) -> None:
|
||
if not self.run_record:
|
||
return
|
||
self.run_record["ended_at"] = _now_iso()
|
||
started_at = self.run_record.get("started_at")
|
||
if isinstance(started_at, str):
|
||
try:
|
||
start_dt = datetime.fromisoformat(started_at)
|
||
self.run_record["duration_s"] = int((datetime.now(timezone.utc) - start_dt).total_seconds())
|
||
except ValueError:
|
||
pass
|
||
if self.current_device_id and not self.run_record.get("device_id"):
|
||
self.run_record["device_id"] = self.current_device_id
|
||
if self.device_id_hint and not self.run_record.get("device_id"):
|
||
self.run_record["device_id"] = self.device_id_hint
|
||
self.run_record["success"] = bool(success)
|
||
safe_record = self._sanitize_run_record(self.run_record)
|
||
path = self._write_run_record(safe_record)
|
||
if path:
|
||
if self.safe_log:
|
||
device_id = safe_record.get("device_id")
|
||
if isinstance(device_id, str) and device_id:
|
||
self.log(f"[OK] 产测记录已保存 (device_id={device_id})。")
|
||
else:
|
||
self.log("[OK] 产测记录已保存。")
|
||
else:
|
||
self.log(f"[OK] 产测记录已保存: {path}")
|
||
self.run_record = None
|
||
|
||
def _on_flash_output(self, line: str) -> None:
|
||
self._parse_flash_progress(line)
|
||
if self.safe_log:
|
||
return
|
||
if line:
|
||
self.log(line)
|
||
|
||
def _parse_flash_progress(self, line: str) -> None:
|
||
if not line:
|
||
return
|
||
if "Wrote " in line and "bytes" in line:
|
||
self.set_progress(100.0)
|
||
if self.active_mode_label:
|
||
self.set_status(f"{self.active_mode_label}进行中... 100%")
|
||
return
|
||
match = re.search(r"(\d+(?:\.\d+)?)%", line)
|
||
if not match:
|
||
return
|
||
try:
|
||
percent = float(match.group(1))
|
||
except ValueError:
|
||
return
|
||
self.set_progress(percent)
|
||
if self.active_mode_label:
|
||
self.set_status(f"{self.active_mode_label}进行中... {percent:.0f}%")
|
||
|
||
def set_status(self, message: str) -> None:
|
||
self.root.after(0, lambda: self.status_var.set(message))
|
||
|
||
def set_status_tone(self, tone: str) -> None:
|
||
palette = {
|
||
"info": "#1a73e8",
|
||
"success": "#0b8a3c",
|
||
"warn": "#c77c02",
|
||
"error": "#c0362c",
|
||
}
|
||
color = palette.get(tone, "#000")
|
||
self.root.after(0, lambda: self.status_label.configure(foreground=color))
|
||
|
||
def set_busy_indicator(self, busy: bool, mode: str = "indeterminate") -> None:
|
||
def _apply() -> None:
|
||
if busy:
|
||
self.progress.pack(side="right", padx=(8, 0))
|
||
if mode == "determinate":
|
||
self.progress.configure(mode="determinate")
|
||
self.progress_value.set(0.0)
|
||
else:
|
||
self.progress.configure(mode="indeterminate")
|
||
self.progress.start(8)
|
||
return
|
||
self.progress.stop()
|
||
self.progress.pack_forget()
|
||
|
||
self.root.after(0, _apply)
|
||
|
||
def set_progress(self, value: float) -> None:
|
||
def _apply() -> None:
|
||
self.progress_value.set(max(0.0, min(100.0, value)))
|
||
|
||
self.root.after(0, _apply)
|
||
|
||
def clear_log(self, update_status: bool = True) -> None:
|
||
self.log_box.configure(state="normal")
|
||
self.log_box.delete("1.0", "end")
|
||
self.log_box.configure(state="disabled")
|
||
if update_status:
|
||
self.set_status("日志已清空。")
|
||
self.set_status_tone("info")
|
||
|
||
def copy_log(self) -> None:
|
||
content = self.log_box.get("1.0", "end").strip()
|
||
self.root.clipboard_clear()
|
||
if content:
|
||
self.root.clipboard_append(content)
|
||
self.set_status("日志已复制到剪贴板。")
|
||
self.set_status_tone("success")
|
||
else:
|
||
self.set_status("日志为空。")
|
||
self.set_status_tone("warn")
|
||
|
||
def set_hint_text(self, text: str, tone: str = "info") -> None:
|
||
palette = {
|
||
"info": "#555555",
|
||
"success": "#0b8a3c",
|
||
"warn": "#c77c02",
|
||
"error": "#c0362c",
|
||
}
|
||
color = palette.get(tone, "#555555")
|
||
|
||
def _apply() -> None:
|
||
self.hint_var.set(text)
|
||
self.hint_label.configure(foreground=color)
|
||
|
||
self.root.after(0, _apply)
|
||
|
||
def _normalize_port(self, port: Optional[str]) -> str:
|
||
if not port:
|
||
return ""
|
||
return port.strip().upper()
|
||
|
||
def _on_port_change(self, *_args) -> None:
|
||
port = self._normalize_port(self.port_var.get())
|
||
if port == self._last_port_value:
|
||
return
|
||
self._last_port_value = port
|
||
if self.is_busy:
|
||
self.update_flash_buttons()
|
||
return
|
||
|
||
if not port:
|
||
self.checked_port = None
|
||
self.summary_attempted = False
|
||
self.summary_ok = False
|
||
self.last_rec_mode = None
|
||
self.secure_boot_blocked = False
|
||
self.secure_boot_enabled = None
|
||
self.boot_chain_present = None
|
||
self.boot_chain_probe_reason = None
|
||
self.download_mode_disabled = None
|
||
self.secure_download_enabled = None
|
||
self.manual_encrypt_disabled = None
|
||
self.device_id_hint = None
|
||
self.recommend_var.set("请选择串口后点击开始。")
|
||
self.set_hint_text("", tone="info")
|
||
self.card_port_var.set("未选择")
|
||
self.card_secure_boot_var.set("未检测")
|
||
self.card_flash_enc_var.set("未检测")
|
||
self.card_recommend_var.set("待检测")
|
||
self.update_flash_buttons()
|
||
return
|
||
|
||
if self.summary_attempted:
|
||
self.summary_attempted = False
|
||
self.summary_ok = False
|
||
self.last_rec_mode = None
|
||
self.secure_boot_blocked = False
|
||
self.secure_boot_enabled = None
|
||
self.boot_chain_present = None
|
||
self.boot_chain_probe_reason = None
|
||
self.download_mode_disabled = None
|
||
self.secure_download_enabled = None
|
||
self.manual_encrypt_disabled = None
|
||
self.device_id_hint = None
|
||
self.recommend_var.set("串口已变更，请点击开始重新检测。")
|
||
self.set_hint_text("检测结果仅对当前串口有效。", tone="warn")
|
||
elif not self.summary_attempted:
|
||
self.recommend_var.set("准备开始：点击开始自动检测并执行流程。")
|
||
self.set_hint_text("", tone="info")
|
||
|
||
self.card_port_var.set(port)
|
||
self.card_secure_boot_var.set("待检测")
|
||
self.card_flash_enc_var.set("待检测")
|
||
self.card_recommend_var.set("请先设备检测")
|
||
self.update_flash_buttons()
|
||
|
||
def update_flash_buttons(self) -> None:
|
||
self.root.after(0, self._apply_flash_buttons)
|
||
|
||
def _apply_flash_buttons(self) -> None:
|
||
busy = self.is_busy
|
||
|
||
self.detect_btn.state(["disabled"] if busy else ["!disabled"])
|
||
self.port_entry.configure(state="disabled" if busy else "normal")
|
||
|
||
start_state = ["!disabled"]
|
||
if busy:
|
||
start_state = ["disabled"]
|
||
self.start_btn.state(start_state)
|
||
|
||
def disable_buttons(self, progress_mode: str = "indeterminate") -> None:
|
||
self.is_busy = True
|
||
self.set_busy_indicator(True, mode=progress_mode)
|
||
self.update_flash_buttons()
|
||
|
||
def enable_buttons(self) -> None:
|
||
self.is_busy = False
|
||
self.set_busy_indicator(False)
|
||
self.update_flash_buttons()
|
||
self.active_mode_label = ""
|
||
|
||
def on_auto_detect(self) -> None:
|
||
port = self._auto_detect_port(show_dialog=True)
|
||
if port:
|
||
self.port_var.set(port)
|
||
|
||
def on_start(self) -> None:
|
||
self.start_task(self.flow_task, progress_mode="determinate")
|
||
|
||
def start_task(self, func, progress_mode: str = "indeterminate") -> None:
|
||
if self.worker and self.worker.is_alive():
|
||
return
|
||
self.disable_buttons(progress_mode=progress_mode)
|
||
self.set_status("处理中...")
|
||
self.set_status_tone("info")
|
||
self.worker = threading.Thread(target=self._run_wrapper, args=(func,), daemon=True)
|
||
self.worker.start()
|
||
|
||
def _run_wrapper(self, func) -> None:
|
||
try:
|
||
func()
|
||
finally:
|
||
self.root.after(0, self.enable_buttons)
|
||
if self.suppress_ready_status:
|
||
self.suppress_ready_status = False
|
||
return
|
||
self.set_status("就绪。")
|
||
self.set_status_tone("info")
|
||
|
||
def prepare_secure_entries(
|
||
self, entries: Iterable[Tuple[object, Path]]
|
||
) -> Optional[list[FlashSource]]:
|
||
missing: list[Path] = []
|
||
sources: list[Tuple[int, Path, Path]] = []
|
||
|
||
for offset, path in entries:
|
||
source = Path(str(path) + ".enc")
|
||
try:
|
||
offset_int = _parse_offset(offset)
|
||
except ValueError as exc:
|
||
self.log("[错误] 固件偏移配置无效。")
|
||
if not self.safe_log:
|
||
self.log(f" - {exc}")
|
||
self._notify("error", "偏移无效", "固件偏移配置无效，请查看日志。")
|
||
return None
|
||
sources.append((offset_int, path, source))
|
||
if not source.exists():
|
||
missing.append(source)
|
||
|
||
if missing:
|
||
self.log("[错误] 缺少必要固件文件。")
|
||
if not self.safe_log:
|
||
for item in missing:
|
||
self.log(f" - {_display_path(self.base_dir, item)}")
|
||
self._notify("error", "文件缺失", "缺少必要固件文件，请查看日志。")
|
||
return None
|
||
|
||
try:
|
||
load_firmware_decrypt_key()
|
||
except FirmwareEnvelopeError as exc:
|
||
self.log("[错误] .enc 固件解密密钥不可用。")
|
||
if not self.safe_log:
|
||
self.log(f" - {exc}")
|
||
self._notify("error", "缺少解密密钥", str(exc))
|
||
return None
|
||
|
||
resolved: list[FlashSource] = []
|
||
for offset_int, orig, source in sources:
|
||
try:
|
||
verify_file_signature(source)
|
||
except SignatureError as exc:
|
||
self.log("[错误] 固件校验失败。")
|
||
if not self.safe_log:
|
||
self.log(f" - {_display_path(self.base_dir, source)}")
|
||
self.log(f" - {exc}")
|
||
self._notify("error", "文件校验失败", "文件校验失败，请查看日志。")
|
||
return None
|
||
resolved.append(
|
||
FlashSource(
|
||
offset=offset_int,
|
||
path=source,
|
||
name=orig.name,
|
||
)
|
||
)
|
||
|
||
return resolved
|
||
|
||
def _is_force_bootloader_entry(self, entry: FlashSource) -> bool:
|
||
name = entry.name.strip().lower()
|
||
return entry.offset == 0x0 and "bootloader" in name and name.endswith(".bin")
|
||
|
||
@staticmethod
|
||
def _normalize_chip_name(value: object) -> str:
|
||
text = str(value or "").strip().lower()
|
||
return re.sub(r"[^0-9a-z]", "", text)
|
||
|
||
@staticmethod
|
||
def _format_rev_full(rev: int) -> str:
|
||
if rev < 0:
|
||
return "unknown"
|
||
return f"v{rev // 100}.{rev % 100}"
|
||
|
||
def _guard_force_bootloader_write(
|
||
self,
|
||
esp: Any,
|
||
entry: FlashSource,
|
||
payload: bytes,
|
||
*,
|
||
encrypt: bool,
|
||
) -> None:
|
||
from esptool.bin_image import LoadFirmwareImage
|
||
|
||
if not self._is_force_bootloader_entry(entry):
|
||
raise RuntimeError(
|
||
f"force 风险拦截：仅允许 bootloader@0x0，当前条目为 {entry.name} @ {hex(entry.offset)}"
|
||
)
|
||
|
||
expected_chip = self._normalize_chip_name(self.private_cfg.chip)
|
||
chip_name = str(getattr(esp, "CHIP_NAME", "") or "")
|
||
actual_chip = self._normalize_chip_name(chip_name)
|
||
if expected_chip and expected_chip != "auto" and actual_chip and expected_chip != actual_chip:
|
||
raise RuntimeError(
|
||
"force 风险拦截：设备芯片与配置不一致，"
|
||
f"cfg={self.private_cfg.chip}, detected={chip_name or 'unknown'}"
|
||
)
|
||
|
||
try:
|
||
image = LoadFirmwareImage(chip_name or self.private_cfg.chip, payload)
|
||
except Exception as exc: # noqa: BLE001
|
||
raise RuntimeError(
|
||
f"force 风险拦截：无法解析 bootloader 镜像头（{entry.name}），已拒绝写入。"
|
||
) from exc
|
||
|
||
image_chip_id = getattr(image, "chip_id", None)
|
||
target_chip_id = getattr(esp, "IMAGE_CHIP_ID", None)
|
||
if image_chip_id != target_chip_id:
|
||
raise RuntimeError(
|
||
"force 风险拦截：bootloader 芯片 ID 与目标芯片不匹配，"
|
||
f"image={image_chip_id}, target={target_chip_id}"
|
||
)
|
||
|
||
def _to_int(value: object, default: int = 0) -> int:
|
||
try:
|
||
return int(value)
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
image_min_rev = _to_int(getattr(image, "min_rev", 0), 0)
|
||
image_min_rev_full = _to_int(getattr(image, "min_rev_full", 0), 0)
|
||
image_max_rev_full = _to_int(getattr(image, "max_rev_full", 0), 0)
|
||
|
||
if image_max_rev_full == 0:
|
||
use_rev_full_fields = False
|
||
elif image_max_rev_full == 65535:
|
||
use_rev_full_fields = not (image_min_rev_full == 0 and image_min_rev != 0)
|
||
else:
|
||
use_rev_full_fields = True
|
||
|
||
if use_rev_full_fields:
|
||
try:
|
||
chip_rev_full = _to_int(getattr(esp, "get_chip_revision")(), -1)
|
||
except Exception as exc: # noqa: BLE001
|
||
raise RuntimeError("force 风险拦截：无法读取芯片 revision 信息。") from exc
|
||
if chip_rev_full < image_min_rev_full or chip_rev_full > image_max_rev_full:
|
||
upper = (
|
||
"max rev not set"
|
||
if image_max_rev_full == 65535
|
||
else self._format_rev_full(image_max_rev_full)
|
||
)
|
||
raise RuntimeError(
|
||
"force 风险拦截：bootloader revision 约束不满足，"
|
||
f"requires=[{self._format_rev_full(image_min_rev_full)} - {upper}], "
|
||
f"chip={self._format_rev_full(chip_rev_full)}"
|
||
)
|
||
else:
|
||
try:
|
||
if chip_name == "ESP32-C3":
|
||
chip_rev = _to_int(getattr(esp, "get_minor_chip_version")(), -1)
|
||
else:
|
||
chip_rev = _to_int(getattr(esp, "get_major_chip_version")(), -1)
|
||
except Exception as exc: # noqa: BLE001
|
||
raise RuntimeError("force 风险拦截：无法读取芯片 revision 信息。") from exc
|
||
if chip_rev < image_min_rev:
|
||
raise RuntimeError(
|
||
"force 风险拦截：bootloader 最小 revision 要求不满足，"
|
||
f"requires={image_min_rev}, chip={chip_rev}"
|
||
)
|
||
|
||
if not encrypt and chip_name != "ESP8266":
|
||
try:
|
||
flash_encryption_enabled = bool(getattr(esp, "get_flash_encryption_enabled")())
|
||
except Exception as exc: # noqa: BLE001
|
||
raise RuntimeError(
|
||
"force 风险拦截：无法确认 Flash Encryption 状态，已拒绝明文 force 写入。"
|
||
) from exc
|
||
if flash_encryption_enabled:
|
||
raise RuntimeError(
|
||
"force 风险拦截：检测到 Flash Encryption 已启用，拒绝明文 force 写入。"
|
||
)
|
||
|
||
secure_download_mode = bool(getattr(esp, "secure_download_mode", False))
|
||
if chip_name != "ESP32" and secure_download_mode:
|
||
try:
|
||
security_info = getattr(esp, "get_security_info")()
|
||
flash_crypt_cnt = _to_int(
|
||
security_info.get("flash_crypt_cnt") if isinstance(security_info, dict) else 0
|
||
)
|
||
except Exception as exc: # noqa: BLE001
|
||
raise RuntimeError(
|
||
"force 风险拦截：无法读取 secure download 状态下的 flash_crypt_cnt。"
|
||
) from exc
|
||
if bin(flash_crypt_cnt).count("1") & 1 != 0:
|
||
raise RuntimeError(
|
||
"force 风险拦截：检测到加密状态，拒绝在 secure download 模式下明文 force 写入。"
|
||
)
|
||
|
||
def ensure_files_exist(self, paths: Iterable[Path]) -> bool:
|
||
missing = [str(p) for p in paths if not p.exists()]
|
||
if missing:
|
||
self.log("[错误] 缺少必要文件。")
|
||
if not self.safe_log:
|
||
for item in missing:
|
||
self.log(f" - {_display_path(self.base_dir, Path(item))}")
|
||
self._notify("error", "文件缺失", "缺少必要文件，请查看日志。")
|
||
return False
|
||
return True
|
||
|
||
def _auto_detect_port(self, show_dialog: bool) -> Optional[str]:
|
||
ports = list_serial_ports(self.python_cmd, show_dialog)
|
||
if not ports:
|
||
if show_dialog:
|
||
self._notify("warn", "未检测到串口", "未检测到可用串口，请手动输入。")
|
||
return None
|
||
selected = pick_preferred_port(ports)
|
||
if len(ports) > 1:
|
||
if self.safe_log:
|
||
self.log("[提示] 检测到多个串口，已自动选择。")
|
||
else:
|
||
self.log(f"[提示] 检测到多个串口: {', '.join(ports)}")
|
||
self.log(f"[提示] 自动选择: {selected}")
|
||
if show_dialog:
|
||
message = "检测到多个串口，已自动选择。"
|
||
if not self.safe_log:
|
||
message = f"检测到多个串口:\n{', '.join(ports)}\n\n已自动选择: {selected}"
|
||
self._notify("info", "检测到多个串口", message)
|
||
return selected
|
||
|
||
def _auto_set_port(self, initial: bool) -> None:
|
||
if not initial and self.port_var.get().strip():
|
||
return
|
||
port = self._auto_detect_port(show_dialog=False)
|
||
if port:
|
||
self.port_var.set(port)
|
||
|
||
def preflight_check(self) -> None:
|
||
if not self.ensure_tools_available():
|
||
self.set_status("缺少工具。")
|
||
self.set_status_tone("error")
|
||
return
|
||
self.log("[OK] 检测到工具，准备就绪。")
|
||
self._refresh_mode_status()
|
||
self._refresh_key_status()
|
||
self.set_status("就绪。")
|
||
self.set_status_tone("success")
|
||
self.update_flash_buttons()
|
||
|
||
def flow_task(self) -> None:
|
||
self.suppress_ready_status = True
|
||
self.current_device_id = None
|
||
self.last_flash_app_only = None
|
||
self._post_flash_reset_done = False
|
||
self._begin_run_record("")
|
||
success = False
|
||
try:
|
||
port = self.port_var.get().strip()
|
||
if not port:
|
||
port = self._auto_detect_port(show_dialog=False) or ""
|
||
if not port:
|
||
self.set_status("未检测到串口。")
|
||
self.set_status_tone("warn")
|
||
self._notify("warn", "未检测到串口", "未检测到可用串口，请手动输入。")
|
||
self._record_error("preflight", "no_port")
|
||
return
|
||
self.port_var.set(port)
|
||
if self.run_record is not None:
|
||
self.run_record["port"] = port
|
||
|
||
if not self._ensure_private_config(port, show_dialog=True):
|
||
self.set_status("配置加载失败。")
|
||
self.set_status_tone("error")
|
||
self._record_error("preflight", "private_config_missing")
|
||
return
|
||
|
||
self.set_progress(0.0)
|
||
self.recommend_var.set("步骤 1/3：设备检测中...")
|
||
if not self.device_check_task():
|
||
self._record_error("device_check", "failed")
|
||
return
|
||
|
||
flash_start_ts = time.time()
|
||
mode = self._decide_flash_mode()
|
||
if mode is None:
|
||
reason = self.last_flash_block_reason or "mode_unknown"
|
||
self._record_error("flash", reason)
|
||
self._record_step_result(
|
||
"flash",
|
||
flash_start_ts,
|
||
"failed",
|
||
{"reason": reason},
|
||
)
|
||
return
|
||
encrypt, app_only, mode_label = mode
|
||
self.last_flash_app_only = app_only
|
||
|
||
self.recommend_var.set(f"步骤 2/3：{mode_label}烧录中...")
|
||
if not self.flash_task(encrypt=encrypt, app_only=app_only):
|
||
self._record_error("flash", "failed")
|
||
return
|
||
|
||
self.recommend_var.set("步骤 3/3：安全复检中...")
|
||
if not self.post_flash_check_task(port):
|
||
self._record_error("post_flash_check", "failed")
|
||
return
|
||
self._attempt_reset(port)
|
||
time.sleep(1.0)
|
||
|
||
self.recommend_var.set("完成")
|
||
self.set_status("流程完成。")
|
||
self.set_status_tone("success")
|
||
self._notify("info", "完成", "烧录流程完成。")
|
||
success = True
|
||
finally:
|
||
self._finalize_run_record(success)
|
||
|
||
def _decide_flash_mode(self) -> Optional[Tuple[bool, bool, str]]:
|
||
self.last_flash_block_reason = None
|
||
if not self.summary_ok:
|
||
self._notify("error", "检测失败", "设备检测未通过，无法继续。")
|
||
self.last_flash_block_reason = "summary_failed"
|
||
return None
|
||
if self.download_mode_disabled is True:
|
||
self._notify("error", "下载模式已禁用", "ROM 下载模式已禁用，无法串口烧录。")
|
||
self.last_flash_block_reason = "download_mode_disabled"
|
||
return None
|
||
if self.secure_download_enabled is True:
|
||
self._notify(
|
||
"error",
|
||
"安全下载模式",
|
||
"检测到安全下载模式已启用，当前工具不支持安全下载协议。",
|
||
)
|
||
self.last_flash_block_reason = "secure_download_enabled"
|
||
return None
|
||
encrypt: Optional[bool] = None
|
||
if self.last_rec_mode == "1":
|
||
encrypt = False
|
||
elif self.last_rec_mode == "2":
|
||
encrypt = True
|
||
if encrypt is None:
|
||
self._notify("error", "模式未知", "无法判断烧录模式，请检查设备状态。")
|
||
self.last_flash_block_reason = "rec_mode_unknown"
|
||
return None
|
||
if encrypt and self.manual_encrypt_disabled is True:
|
||
self._notify(
|
||
"error",
|
||
"串口加密受限",
|
||
"检测到禁用串口手动加密，无法使用当前工具进行加密写入。",
|
||
)
|
||
self.last_flash_block_reason = "manual_encrypt_disabled"
|
||
return None
|
||
if self.secure_boot_enabled is True:
|
||
if self.boot_chain_present is False:
|
||
if encrypt:
|
||
return True, False, "加密全量（初次写入）"
|
||
return False, False, f"{self.plain_text}（初次写入）"
|
||
mode_label = self.app_text if encrypt else f"{self.app_text}（不加密）"
|
||
return encrypt, True, mode_label
|
||
if encrypt is False:
|
||
return False, False, f"{self.plain_text}"
|
||
if encrypt is True:
|
||
return True, False, "加密全量"
|
||
self._notify("error", "模式未知", "无法判断烧录模式，请检查设备状态。")
|
||
return None
|
||
|
||
def _read_efuse_summary(self, port: str) -> Tuple[int, str, Optional[dict]]:
|
||
args = [
|
||
"--chip",
|
||
self.private_cfg.chip,
|
||
"--port",
|
||
port,
|
||
"summary",
|
||
"--format",
|
||
"json",
|
||
]
|
||
code, output = self.run_tool("espefuse", args, stream=False)
|
||
summary_json = _extract_json_blob(output) if code == 0 else None
|
||
return code, output, summary_json
|
||
|
||
def _log_efuse_summary(self, output: str) -> None:
|
||
if not output:
|
||
return
|
||
self.log_section("eFuse Summary")
|
||
if self.safe_log:
|
||
self.log("已获取（内容已脱敏）。", timestamp=False)
|
||
return
|
||
for line in output.splitlines():
|
||
self.log(line, timestamp=False)
|
||
|
||
def device_check_task(self) -> bool:
|
||
start_ts = time.time()
|
||
port = self.port_var.get().strip()
|
||
if not port:
|
||
port = self._auto_detect_port(show_dialog=False) or ""
|
||
if not port:
|
||
self._notify("warn", "未检测到串口", "未检测到可用串口，请手动输入。")
|
||
self._record_step_result(
|
||
"device_check",
|
||
start_ts,
|
||
"failed",
|
||
{"reason": "no_port"},
|
||
)
|
||
return False
|
||
self.port_var.set(port)
|
||
if not self._ensure_private_config(port, show_dialog=True):
|
||
self._record_step_result(
|
||
"device_check",
|
||
start_ts,
|
||
"failed",
|
||
{"reason": "private_config_missing"},
|
||
)
|
||
return False
|
||
if not self.ensure_tools_available():
|
||
self._record_step_result(
|
||
"device_check",
|
||
start_ts,
|
||
"failed",
|
||
{"reason": "tools_missing"},
|
||
)
|
||
return False
|
||
|
||
self.set_status("正在检测设备...")
|
||
self.set_status_tone("info")
|
||
self.log_section("设备检测")
|
||
if not self.safe_log:
|
||
self.log(f"串口: {port}")
|
||
self.recommend_var.set("正在检测设备...")
|
||
self.set_hint_text("", tone="info")
|
||
|
||
code, summary_output, summary_json = self._read_efuse_summary(port)
|
||
if summary_output:
|
||
self._log_efuse_summary(summary_output)
|
||
|
||
self.summary_attempted = True
|
||
if code != 0 or summary_json is None:
|
||
if code != 0:
|
||
self.log("[警告] 设备检测失败。")
|
||
else:
|
||
self.log("[警告] 设备检测失败：JSON 解析失败。")
|
||
self.set_status("检测失败。")
|
||
self.set_status_tone("warn")
|
||
self.summary_ok = False
|
||
self.last_rec_mode = None
|
||
self.secure_boot_blocked = False
|
||
self.secure_boot_enabled = None
|
||
self.boot_chain_present = None
|
||
self.boot_chain_probe_reason = None
|
||
self.download_mode_disabled = None
|
||
self.secure_download_enabled = None
|
||
self.manual_encrypt_disabled = None
|
||
self.checked_port = None
|
||
self.card_secure_boot_var.set("检测失败")
|
||
self.card_flash_enc_var.set("检测失败")
|
||
self.card_recommend_var.set("请检查连接后重试")
|
||
hints = collect_error_hints(summary_output)
|
||
# 不展示检测结果，只提示信息
|
||
self.recommend_var.set("检测失败，请查看提示后重试。")
|
||
self.set_hint_text("\n".join(f"- {hint}" for hint in hints), tone="warn")
|
||
self.log("操作指引:", timestamp=False)
|
||
for hint in hints:
|
||
self.log(f" - {hint}", timestamp=False)
|
||
self.update_flash_buttons()
|
||
self.log("流程已停止，请重试设备检测。")
|
||
self._record_step_result(
|
||
"device_check",
|
||
start_ts,
|
||
"failed",
|
||
{
|
||
"reason": "summary_json_failed" if summary_json is None else "summary_failed",
|
||
"hints": hints,
|
||
"port": port,
|
||
},
|
||
)
|
||
return False
|
||
rec_mode, rec_reason = recommend_mode_from_json(
|
||
summary_json, self.private_cfg.device_check.crypt_cnt_fields
|
||
)
|
||
self.summary_ok = True
|
||
self.last_rec_mode = rec_mode
|
||
self.secure_boot_blocked = False
|
||
self.secure_boot_enabled = get_efuse_flag(
|
||
summary_json, self.private_cfg.device_check.secure_boot_field
|
||
)
|
||
self.download_mode_disabled = get_efuse_flag(
|
||
summary_json, "DIS_DOWNLOAD_MODE", "UART_DOWNLOAD_DIS"
|
||
)
|
||
self.secure_download_enabled = get_efuse_flag(
|
||
summary_json, "ENABLE_SECURITY_DOWNLOAD", "SECURE_DOWNLOAD_MODE"
|
||
)
|
||
self.manual_encrypt_disabled = get_efuse_flag(
|
||
summary_json, "DIS_DOWNLOAD_MANUAL_ENCRYPT"
|
||
)
|
||
if self.secure_boot_enabled is True and rec_mode == "1":
|
||
self.boot_chain_present, self.boot_chain_probe_reason = self._probe_boot_chain_header(port)
|
||
elif self.secure_boot_enabled is True and rec_mode == "2":
|
||
# When flash encryption is enabled, raw flash reads are ciphertext and
|
||
# cannot be used to infer plaintext image headers at 0x0.
|
||
self.boot_chain_present = True
|
||
self.boot_chain_probe_reason = "skip_probe_flash_encryption_enabled"
|
||
else:
|
||
self.boot_chain_present = None
|
||
self.boot_chain_probe_reason = "skip_probe_not_applicable"
|
||
self.checked_port = port
|
||
mac = self._read_chip_mac(port)
|
||
if mac:
|
||
self.device_id_hint = mac
|
||
if self.run_record and not self.run_record.get("device_id"):
|
||
self.run_record["device_id"] = mac
|
||
flash_enc_status = "未知"
|
||
if rec_mode == "1":
|
||
flash_enc_status = "未启用"
|
||
elif rec_mode == "2":
|
||
flash_enc_status = "已启用"
|
||
self.card_flash_enc_var.set(flash_enc_status)
|
||
|
||
if self.secure_boot_enabled is True:
|
||
self.card_secure_boot_var.set("已启用")
|
||
elif self.secure_boot_enabled is False:
|
||
self.card_secure_boot_var.set("未启用")
|
||
else:
|
||
self.card_secure_boot_var.set("未知")
|
||
|
||
boot_chain_missing = self.boot_chain_present is False
|
||
|
||
if self.secure_boot_enabled:
|
||
if boot_chain_missing:
|
||
if rec_mode == "1":
|
||
recommend_text = f"下一步：{self.plain_text}（初次全量）"
|
||
elif rec_mode == "2":
|
||
recommend_text = "检测到安全启动已启用且启动链缺失，将执行加密全量初次写入。"
|
||
else:
|
||
recommend_text = "检测到安全启动已启用且启动链缺失，请先核对设备信息。"
|
||
elif rec_mode == "1":
|
||
recommend_text = f"下一步：{self.app_text}（不加密）"
|
||
elif rec_mode == "2":
|
||
recommend_text = f"下一步：{self.app_text}（加密）"
|
||
else:
|
||
recommend_text = "请先核对设备信息"
|
||
elif rec_mode == "1":
|
||
recommend_text = f"下一步：{self.plain_text}"
|
||
elif rec_mode == "2":
|
||
recommend_text = "检测到闪存加密已启用但安全启动未启用，将执行加密全量烧录并在首启后启用安全启动。"
|
||
else:
|
||
recommend_text = "请先核对设备信息"
|
||
|
||
if self.download_mode_disabled is True:
|
||
recommend_text = "检测到下载模式已禁用，请按工艺处理。"
|
||
elif self.secure_download_enabled is True:
|
||
recommend_text = "检测到安全下载模式已启用，请按工艺处理。"
|
||
elif self.manual_encrypt_disabled is True and rec_mode == "2":
|
||
recommend_text = "检测到禁用串口手动加密，请使用预加密固件或 OTA。"
|
||
|
||
self.recommend_var.set(recommend_text)
|
||
self.card_recommend_var.set(recommend_text)
|
||
hint_lines = []
|
||
hint_tone = "info"
|
||
if rec_mode == "1":
|
||
if self.secure_boot_enabled:
|
||
if boot_chain_missing:
|
||
hint_lines.append("检测到启动链缺失，将执行模式 A（全量写入）完成初次写入。")
|
||
else:
|
||
hint_lines.append("提示：安全启动已启用但闪存加密未启用，将以非加密写入方式执行模式 C。")
|
||
hint_tone = "warn"
|
||
else:
|
||
hint_lines.append("已就绪：执行模式 A（全量写入）。")
|
||
elif rec_mode == "2":
|
||
if self.secure_boot_enabled:
|
||
if boot_chain_missing:
|
||
hint_lines.append("检测到启动链缺失，将执行加密全量初次写入。")
|
||
hint_tone = "warn"
|
||
else:
|
||
hint_lines.append("已就绪：执行模式 C（加密）。")
|
||
else:
|
||
hint_lines.append("提示：将执行加密全量烧录，首启后自动启用安全启动。")
|
||
hint_tone = "warn"
|
||
else:
|
||
hint_lines.append("信息不明确：请按工艺要求核对后再选择模式。")
|
||
hint_tone = "warn"
|
||
|
||
if self.secure_boot_enabled:
|
||
if boot_chain_missing:
|
||
hint_lines.append("检测到 flash@0x0 启动头无效/空白，将改为全量写入并在需要时使用 force。")
|
||
else:
|
||
hint_lines.append("已限制，仅允许模式 C。")
|
||
|
||
if self.download_mode_disabled is True:
|
||
hint_lines.append("检测到下载模式已禁用，串口烧录不可用。")
|
||
hint_tone = "error"
|
||
if self.secure_download_enabled is True:
|
||
hint_lines.append("检测到安全下载模式已启用，当前工具不支持安全下载。")
|
||
if hint_tone != "error":
|
||
hint_tone = "warn"
|
||
if self.manual_encrypt_disabled is True:
|
||
if rec_mode == "2":
|
||
hint_lines.append("检测到禁用串口手动加密，无法进行加密写入。")
|
||
hint_tone = "error"
|
||
else:
|
||
hint_lines.append("检测到禁用串口手动加密（仅影响加密写入）。")
|
||
if hint_tone != "error":
|
||
hint_tone = "warn"
|
||
|
||
self.set_hint_text("\n".join(hint_lines), tone=hint_tone)
|
||
self.log("-" * 46, timestamp=False)
|
||
self.log("设备检测完成，已自动确定烧录模式。")
|
||
self.set_status("检测完成。")
|
||
self.set_status_tone("success")
|
||
self.update_flash_buttons()
|
||
self._record_step_result(
|
||
"device_check",
|
||
start_ts,
|
||
"ok",
|
||
{
|
||
"port": port,
|
||
"recommend_mode": rec_mode,
|
||
"recommend_reason": rec_reason,
|
||
"secure_boot_enabled": self.secure_boot_enabled,
|
||
"boot_chain_present": self.boot_chain_present,
|
||
"boot_chain_probe_reason": self.boot_chain_probe_reason,
|
||
"flash_enc_status": flash_enc_status,
|
||
"download_mode_disabled": self.download_mode_disabled,
|
||
"secure_download_enabled": self.secure_download_enabled,
|
||
"manual_encrypt_disabled": self.manual_encrypt_disabled,
|
||
},
|
||
)
|
||
return True
|
||
|
||
def _format_recommendation(
|
||
self, mode: Optional[str], reason: Optional[str], warn_reason: Optional[str]
|
||
) -> str:
|
||
if warn_reason and not mode:
|
||
return f"摘要不可用（{warn_reason}）。"
|
||
if mode:
|
||
if mode == "1":
|
||
label = self.plain_text
|
||
elif mode == "2":
|
||
label = self.app_text
|
||
else:
|
||
label = f"模式 {mode}"
|
||
reason_suffix = f"（{reason}）" if reason else ""
|
||
if mode in ("1", "2"):
|
||
return f"下一步：{label}（模式 {mode}{reason_suffix}）"
|
||
return f"模式提示：{mode}{reason_suffix}"
|
||
return "暂无指引。"
|
||
|
||
def flash_task(self, encrypt: bool, app_only: bool = False) -> bool:
|
||
start_ts = time.time()
|
||
if not self.summary_attempted:
|
||
self.log("[警告] 请先完成设备检测。")
|
||
self._notify("warn", "请先设备检测", "请先完成设备检测，再开始烧录。")
|
||
self._record_step_result(
|
||
"flash",
|
||
start_ts,
|
||
"failed",
|
||
{"reason": "summary_missing"},
|
||
)
|
||
return False
|
||
|
||
port = self.port_var.get().strip()
|
||
if not port:
|
||
port = self._auto_detect_port(show_dialog=False) or ""
|
||
if not port:
|
||
self._notify("warn", "未检测到串口", "未检测到可用串口，请手动输入。")
|
||
self._record_step_result(
|
||
"flash",
|
||
start_ts,
|
||
"failed",
|
||
{"reason": "no_port"},
|
||
)
|
||
return False
|
||
self.port_var.set(port)
|
||
if not self._ensure_private_config(port, show_dialog=True):
|
||
self._record_step_result(
|
||
"flash",
|
||
start_ts,
|
||
"failed",
|
||
{"reason": "private_config_missing"},
|
||
)
|
||
return False
|
||
bin_dir = select_bin_dir(self.base_dir, self.private_cfg.bin_dir)
|
||
entries = build_flash_entries(
|
||
bin_dir,
|
||
self.private_cfg.layout,
|
||
app_only,
|
||
self.private_cfg.app_partition_offset,
|
||
self.private_cfg.app_bin_name,
|
||
)
|
||
secure_entries = self.prepare_secure_entries(entries)
|
||
if not secure_entries:
|
||
return False
|
||
self.log("固件目录已加载。")
|
||
|
||
if app_only:
|
||
mode_label = self.app_text if encrypt else f"{self.app_text}（不加密）"
|
||
else:
|
||
mode_label = self.plain_text
|
||
self.active_mode_label = mode_label
|
||
self.suppress_ready_status = True
|
||
self.set_progress(0.0)
|
||
self.set_status(f"{mode_label}进行中... 0%")
|
||
self.set_status_tone("info")
|
||
self.log_section(f"{mode_label}烧录开始")
|
||
if not self.safe_log:
|
||
self.log(f"串口: {port} | 模式: {mode_label}")
|
||
|
||
# Force is only allowed for secure-boot recovery of a missing boot chain.
|
||
force_full_flash = bool(
|
||
self.secure_boot_enabled is True and not app_only and self.boot_chain_present is False
|
||
)
|
||
if force_full_flash:
|
||
low_entries = [entry for entry in secure_entries if entry.offset < 0x8000]
|
||
force_entries = [entry for entry in low_entries if self._is_force_bootloader_entry(entry)]
|
||
if len(low_entries) != 1 or len(force_entries) != 1:
|
||
self.log("[错误] force 安全策略校验失败：仅允许 bootloader@0x0 使用 force。")
|
||
self._record_step_result(
|
||
"flash",
|
||
start_ts,
|
||
"failed",
|
||
{
|
||
"reason": "force_policy_mismatch",
|
||
"entries": [
|
||
{"name": entry.name, "offset": entry.offset, "path": str(entry.path)}
|
||
for entry in secure_entries
|
||
],
|
||
},
|
||
)
|
||
self._notify("error", "烧录策略异常", "force 策略校验失败，已停止以避免误写低地址。")
|
||
return False
|
||
self.log("[提示] 检测到安全启动已启用且启动链缺失：仅对 bootloader@0x0 启用 force。")
|
||
|
||
strict_flash = bool(encrypt or self.secure_boot_enabled)
|
||
max_attempts = self.flash_secure_max_attempts if strict_flash else self.flash_max_attempts
|
||
attempts_used = 0
|
||
flash_ok = False
|
||
last_output = ""
|
||
last_report: Optional[FlashReport] = None
|
||
last_failure_stage = "固件烧录"
|
||
baud_override: Optional[str] = None
|
||
|
||
for attempt in range(1, max_attempts + 1):
|
||
attempts_used = attempt
|
||
if attempt > 1:
|
||
self.log(f"[提示] 第 {attempt} 次烧录尝试。")
|
||
code, output, report = self.do_flash(
|
||
port,
|
||
encrypt=encrypt,
|
||
entries=secure_entries,
|
||
on_output=self._on_flash_output,
|
||
baud_override=baud_override,
|
||
force=force_full_flash,
|
||
)
|
||
last_output = output
|
||
last_report = report
|
||
if not output.strip():
|
||
self.log("(无输出)")
|
||
if code == 0:
|
||
self.set_status("写后校验中...")
|
||
self.set_status_tone("info")
|
||
verify_ok, verify_err = self._verify_flash_readback(
|
||
port, report, encrypt, baud_override=baud_override
|
||
)
|
||
self._log_flash_report(report)
|
||
if verify_ok:
|
||
self.log("[OK] 固件烧录完成。")
|
||
self.set_progress(100.0)
|
||
flash_ok = True
|
||
break
|
||
last_failure_stage = "写后校验"
|
||
last_output = verify_err or "readback_verify_failed"
|
||
self.log("[错误] 写后校验失败。")
|
||
else:
|
||
if detect_secure_boot_block(output):
|
||
self.secure_boot_blocked = True
|
||
self.update_flash_buttons()
|
||
if strict_flash:
|
||
break
|
||
|
||
if attempt < max_attempts:
|
||
self.log("[警告] 烧录失败，准备复位后重试。")
|
||
if baud_override is None:
|
||
baud_override = "115200"
|
||
self.log("[提示] 已自动降速到 115200 后重试。")
|
||
self._attempt_reset(port)
|
||
time.sleep(1.0)
|
||
|
||
if not flash_ok:
|
||
failure_entries = None
|
||
if last_report:
|
||
failure_entries = [
|
||
{
|
||
"name": item.name,
|
||
"offset": item.offset,
|
||
"size": item.size,
|
||
"sha256": item.sha256,
|
||
"readback_sha256": item.readback_sha256,
|
||
"verified": item.verified,
|
||
}
|
||
for item in last_report.entries
|
||
]
|
||
self._record_step_result(
|
||
"flash",
|
||
start_ts,
|
||
"failed",
|
||
{
|
||
"reason": last_failure_stage,
|
||
"attempts": attempts_used,
|
||
"mode": mode_label,
|
||
"encrypt": encrypt,
|
||
"app_only": app_only,
|
||
"force": force_full_flash,
|
||
"app_bin_name": self.private_cfg.app_bin_name,
|
||
"readback_errors": last_report.readback_errors if last_report else None,
|
||
"entries": failure_entries,
|
||
},
|
||
)
|
||
self.handle_flash_failure(last_failure_stage, last_output)
|
||
return False
|
||
|
||
if last_report:
|
||
app_entry = next(
|
||
(item for item in last_report.entries if item.name == self.private_cfg.app_bin_name),
|
||
None,
|
||
)
|
||
self._record_step_result(
|
||
"flash",
|
||
start_ts,
|
||
"ok",
|
||
{
|
||
"attempts": attempts_used,
|
||
"mode": mode_label,
|
||
"encrypt": encrypt,
|
||
"app_only": app_only,
|
||
"force": force_full_flash,
|
||
"app_bin_name": self.private_cfg.app_bin_name,
|
||
"entries": [
|
||
{
|
||
"name": item.name,
|
||
"offset": item.offset,
|
||
"size": item.size,
|
||
"sha256": item.sha256,
|
||
"readback_sha256": item.readback_sha256,
|
||
"verified": item.verified,
|
||
}
|
||
for item in last_report.entries
|
||
],
|
||
"readback_ok": last_report.readback_ok,
|
||
"readback_errors": last_report.readback_errors,
|
||
"app_sha256": app_entry.sha256 if app_entry else None,
|
||
},
|
||
)
|
||
else:
|
||
self._record_step_result(
|
||
"flash",
|
||
start_ts,
|
||
"ok",
|
||
{
|
||
"attempts": attempts_used,
|
||
"mode": mode_label,
|
||
"encrypt": encrypt,
|
||
"app_only": app_only,
|
||
"force": force_full_flash,
|
||
},
|
||
)
|
||
|
||
self.log("[OK] 烧录完成。")
|
||
needs_reset = self._needs_post_flash_reset()
|
||
if not needs_reset:
|
||
self._post_flash_reset_done = True
|
||
if needs_reset:
|
||
self.log("[提示] 正在复位设备进入应用模式...")
|
||
self._attempt_reset(port)
|
||
self._post_flash_reset_done = True
|
||
time.sleep(1.0)
|
||
self.set_status("烧录完成。")
|
||
self.set_status_tone("success")
|
||
self.set_hint_text("", tone="success")
|
||
return True
|
||
|
||
def post_flash_check_task(self, port: str) -> bool:
|
||
start_ts = time.time()
|
||
if not self.ensure_tools_available():
|
||
self._record_step_result(
|
||
"post_flash_check",
|
||
start_ts,
|
||
"failed",
|
||
{"reason": "tools_missing"},
|
||
)
|
||
return False
|
||
|
||
self.set_status("安全复检中...")
|
||
self.set_status_tone("info")
|
||
self.log_section("安全复检")
|
||
if not self.safe_log:
|
||
self.log(f"串口: {port}")
|
||
|
||
timeout_s = max(5.0, self.ready_timeout_s)
|
||
deadline = time.time() + timeout_s
|
||
attempts = 0
|
||
last_errors: list[str] = []
|
||
last_details: dict = {}
|
||
last_output = ""
|
||
|
||
if not self._post_flash_reset_done:
|
||
self._attempt_reset(port)
|
||
self._post_flash_reset_done = True
|
||
time.sleep(1.0)
|
||
|
||
while time.time() < deadline:
|
||
attempts += 1
|
||
wait_cap = 60.0
|
||
if self.last_flash_app_only:
|
||
wait_cap = 5.0
|
||
wait_s = min(wait_cap, max(1.0, deadline - time.time()))
|
||
self.log(f"[提示] 等待设备完成安全流程（约 {int(wait_s)} 秒）...")
|
||
time.sleep(wait_s)
|
||
code, output, summary_json = self._read_efuse_summary(port)
|
||
last_output = output
|
||
if code == 0 and summary_json is not None:
|
||
ok, details, errors = evaluate_post_flash_status(
|
||
summary_json, self.private_cfg.device_check
|
||
)
|
||
last_details = details
|
||
if ok:
|
||
self.log("[OK] 安全复检通过。")
|
||
self._record_step_result(
|
||
"post_flash_check",
|
||
start_ts,
|
||
"ok",
|
||
{
|
||
"attempts": attempts,
|
||
"details": details,
|
||
},
|
||
)
|
||
self.set_status("安全复检完成。")
|
||
self.set_status_tone("success")
|
||
return True
|
||
last_errors = errors
|
||
else:
|
||
last_errors = [
|
||
"summary_failed" if code != 0 else "summary_json_failed"
|
||
]
|
||
|
||
self.log("[错误] 安全复检失败。")
|
||
if not self.safe_log and last_output:
|
||
self.log(last_output)
|
||
if last_errors:
|
||
self.log(f"[错误] 安全复检未通过: {', '.join(last_errors)}")
|
||
self.set_status("安全复检失败。")
|
||
self.set_status_tone("error")
|
||
if last_errors:
|
||
self.set_hint_text(
|
||
"\n".join(f"- {item}" for item in last_errors), tone="error"
|
||
)
|
||
self._notify("error", "安全复检失败", "安全复检失败，请查看日志。")
|
||
self._record_step_result(
|
||
"post_flash_check",
|
||
start_ts,
|
||
"failed",
|
||
{
|
||
"attempts": attempts,
|
||
"errors": last_errors,
|
||
"details": last_details,
|
||
},
|
||
)
|
||
return False
|
||
|
||
def handle_flash_failure(self, stage: str, output: str) -> None:
|
||
self.log(f"[错误] {stage}失败。请检查连接、串口及 BOOT/RESET 操作。")
|
||
self.set_status(f"{stage}失败。")
|
||
self.set_status_tone("error")
|
||
self.set_progress(0.0)
|
||
if detect_secure_boot_block(output):
|
||
self.secure_boot_blocked = True
|
||
if self.secure_boot_enabled is not True:
|
||
self.secure_boot_enabled = True
|
||
self.card_secure_boot_var.set("已启用")
|
||
self.update_flash_buttons()
|
||
hints = collect_error_hints(output)
|
||
if detect_secure_boot_block(output):
|
||
if self.boot_chain_present is False:
|
||
hints.insert(0, "检测到安全启动限制：启动链缺失时请使用全量写入（必要时 force）。")
|
||
else:
|
||
hints.insert(0, "检测到安全启动已启用，常规更新仅允许模式 C（App-only）。")
|
||
self.set_hint_text("\n".join(f"- {hint}" for hint in hints), tone="error")
|
||
self.log("操作指引:", timestamp=False)
|
||
for hint in hints:
|
||
self.log(f" - {hint}", timestamp=False)
|
||
self._notify("error", f"{stage}失败", f"{stage}失败，请查看日志与提示。")
|
||
|
||
def _attempt_reset(self, port: str) -> None:
|
||
args = ["--chip", self.private_cfg.chip, "--port", port, "run"]
|
||
code, output = self.run_tool("esptool", args, stream=False)
|
||
if code != 0:
|
||
self.log(f"[警告] 复位失败：{output.strip()}")
|
||
|
||
def _needs_post_flash_reset(self) -> bool:
|
||
after = self.private_cfg.flash_args.get("after")
|
||
if after is None:
|
||
return False
|
||
after_norm = str(after).strip()
|
||
if not after_norm:
|
||
return False
|
||
after_norm = after_norm.replace("_", "-").lower()
|
||
return after_norm in ("no-reset", "noreset", "none", "no")
|
||
|
||
def _read_flash_bytes(
|
||
self, port: str, offset: int, size: int, baud_override: Optional[str] = None
|
||
) -> Optional[bytes]:
|
||
fd, temp_path = tempfile.mkstemp(suffix=".bin")
|
||
os.close(fd)
|
||
args = ["--chip", self.private_cfg.chip, "--port", port]
|
||
baud_value = baud_override or self.private_cfg.baud
|
||
if baud_value:
|
||
baud_int = _value_to_int(baud_value)
|
||
if baud_int is not None and baud_int > 921600 and baud_override is None:
|
||
baud_value = "115200"
|
||
args.extend(["--baud", str(baud_value)])
|
||
try:
|
||
read_args = args + ["read-flash", hex(offset), str(size), temp_path]
|
||
code, output = self.run_tool("esptool", read_args, stream=False)
|
||
if code != 0 and ("Unknown command" in output or "invalid choice" in output):
|
||
read_args = args + ["read_flash", hex(offset), str(size), temp_path]
|
||
code, output = self.run_tool("esptool", read_args, stream=False)
|
||
if code != 0:
|
||
self.log(f"[警告] 读取闪存失败：{output.strip()}")
|
||
return None
|
||
try:
|
||
return Path(temp_path).read_bytes()
|
||
except OSError as exc:
|
||
self.log(f"[警告] 读取闪存临时文件失败：{exc}")
|
||
return None
|
||
finally:
|
||
try:
|
||
os.remove(temp_path)
|
||
except OSError:
|
||
pass
|
||
|
||
def _partition_table_offset_from_layout(self) -> Optional[int]:
|
||
for offset, name in self.private_cfg.layout:
|
||
if not isinstance(name, str):
|
||
continue
|
||
lowered = name.lower()
|
||
if "partition-table" not in lowered and "partition_table" not in lowered:
|
||
continue
|
||
try:
|
||
return _parse_offset(offset)
|
||
except ValueError:
|
||
self.log(f"[警告] 分区表偏移无效：{offset}")
|
||
return None
|
||
return None
|
||
|
||
def _probe_boot_chain_header(self, port: str) -> tuple[Optional[bool], Optional[str]]:
|
||
header = self._read_flash_bytes(port, 0x0, 32)
|
||
if header is None:
|
||
self.log("[警告] 启动链检测跳过：读取 flash@0x0 失败。")
|
||
return None, "probe_failed"
|
||
has_boot_chain, reason = detect_boot_chain_header_state(header)
|
||
if not has_boot_chain:
|
||
self.log(f"[警告] 启动链缺失（flash@0x0，{reason}）。")
|
||
return False, reason
|
||
|
||
partition_offset = self._partition_table_offset_from_layout()
|
||
if partition_offset is None:
|
||
self.log("[提示] 未找到分区表偏移，已按 bootloader 头有效处理。")
|
||
return True, "boot_ok_partition_unknown"
|
||
|
||
partition_header = self._read_flash_bytes(port, partition_offset, 4)
|
||
if partition_header is None:
|
||
self.log(f"[警告] 启动链检测跳过：读取分区表头失败（{hex(partition_offset)}）。")
|
||
return None, "partition_probe_failed"
|
||
if all(value == 0xFF for value in partition_header):
|
||
self.log(f"[警告] 启动链缺失（分区表空白，{hex(partition_offset)}）。")
|
||
return False, "partition_blank_ff"
|
||
if partition_header[:2] != b"\xAA\x50":
|
||
self.log(
|
||
f"[警告] 启动链缺失（分区表头无效，{hex(partition_offset)}，magic=0x{partition_header[0]:02x}{partition_header[1]:02x}）。"
|
||
)
|
||
return False, "partition_invalid_magic"
|
||
|
||
self.log("[OK] 启动链检测通过（bootloader/partition table 头有效）。")
|
||
return True, "ok"
|
||
|
||
def _read_flash_segment(
|
||
self, port: str, offset: int, size: int, baud_override: Optional[str] = None
|
||
) -> Optional[str]:
|
||
fd, temp_path = tempfile.mkstemp(suffix=".bin")
|
||
os.close(fd)
|
||
args = ["--chip", self.private_cfg.chip, "--port", port]
|
||
baud_value = baud_override or self.private_cfg.baud
|
||
if baud_value:
|
||
baud_int = _value_to_int(baud_value)
|
||
if baud_int is not None and baud_int > 921600 and baud_override is None:
|
||
baud_value = "115200"
|
||
args.extend(["--baud", str(baud_value)])
|
||
try:
|
||
read_args = args + ["read-flash", hex(offset), str(size), temp_path]
|
||
code, output = self.run_tool("esptool", read_args, stream=False)
|
||
if code != 0 and ("Unknown command" in output or "invalid choice" in output):
|
||
read_args = args + ["read_flash", hex(offset), str(size), temp_path]
|
||
code, output = self.run_tool("esptool", read_args, stream=False)
|
||
if code != 0:
|
||
self.log(f"[错误] 读回校验失败：{output.strip()}")
|
||
return None
|
||
try:
|
||
hasher = hashlib.sha256()
|
||
total = 0
|
||
with open(temp_path, "rb") as handle:
|
||
while True:
|
||
chunk = handle.read(65536)
|
||
if not chunk:
|
||
break
|
||
total += len(chunk)
|
||
hasher.update(chunk)
|
||
except OSError as exc:
|
||
self.log(f"[错误] 读回文件读取失败：{exc}")
|
||
return None
|
||
if total != size:
|
||
self.log(f"[警告] 读回长度异常：{total} != {size}")
|
||
return hasher.hexdigest()
|
||
finally:
|
||
try:
|
||
os.remove(temp_path)
|
||
except OSError:
|
||
pass
|
||
|
||
def _verify_flash_readback(
|
||
self,
|
||
port: str,
|
||
report: FlashReport,
|
||
encrypt: bool,
|
||
baud_override: Optional[str] = None,
|
||
) -> Tuple[bool, str]:
|
||
if encrypt:
|
||
report.readback_ok = True
|
||
report.readback_errors = []
|
||
return True, ""
|
||
# esptool already performs in-band hash verification for plaintext writes.
|
||
# Extra readback can fail due transient port locking on some hosts; keep it opt-in.
|
||
if self.private_cfg.ui.get("strict_readback") is not True:
|
||
report.readback_ok = True
|
||
report.readback_errors = []
|
||
return True, ""
|
||
if not report.entries:
|
||
return True, ""
|
||
ok = True
|
||
errors: list[str] = []
|
||
for entry in report.entries:
|
||
readback_hash = self._read_flash_segment(
|
||
port, entry.offset, entry.size, baud_override=baud_override
|
||
)
|
||
if readback_hash is None:
|
||
ok = False
|
||
errors.append(f"{entry.name}:readback_failed")
|
||
continue
|
||
entry.readback_sha256 = readback_hash
|
||
if encrypt:
|
||
entry.verified = None
|
||
continue
|
||
entry.verified = (entry.readback_sha256 == entry.sha256)
|
||
if entry.verified is False:
|
||
ok = False
|
||
errors.append(f"{entry.name}:hash_mismatch")
|
||
report.readback_ok = ok
|
||
report.readback_errors = errors
|
||
return ok, "; ".join(errors)
|
||
|
||
def _log_flash_report(self, report: FlashReport) -> None:
|
||
if self.safe_log:
|
||
return
|
||
for entry in report.entries:
|
||
self.log(f"[HASH] {entry.name} sha256={entry.sha256}")
|
||
if entry.readback_sha256:
|
||
self.log(f"[READBACK] {entry.name} sha256={entry.readback_sha256}")
|
||
if entry.verified is True:
|
||
self.log(f"[OK] 校验一致: {entry.name}")
|
||
elif entry.verified is False:
|
||
self.log(f"[错误] 校验不一致: {entry.name}")
|
||
|
||
def do_write_flash(
|
||
self,
|
||
port: str,
|
||
entries: Iterable[FlashSource],
|
||
encrypt: bool,
|
||
on_output: Optional[Callable[[str], None]],
|
||
baud_override: Optional[str] = None,
|
||
force: bool = False,
|
||
) -> Tuple[int, str, FlashReport]:
|
||
if not self.ensure_esptool_api_available():
|
||
return 1, "esptool API unavailable", FlashReport(entries=[], readback_ok=False, readback_errors=[])
|
||
|
||
entries_list = list(entries)
|
||
try:
|
||
decrypt_key = load_firmware_decrypt_key()
|
||
except FirmwareEnvelopeError as exc:
|
||
return 1, str(exc), FlashReport(entries=[], readback_ok=False, readback_errors=[])
|
||
|
||
report = FlashReport(entries=[], readback_ok=False, readback_errors=[])
|
||
|
||
def _run() -> None:
|
||
import esptool
|
||
from esptool import cmds
|
||
from esptool.logger import EsptoolLogger, log as esptool_log
|
||
|
||
try:
|
||
EsptoolLogger._set_smart_features(False)
|
||
except Exception:
|
||
pass
|
||
try:
|
||
esptool_log.set_verbosity("normal")
|
||
except Exception:
|
||
pass
|
||
|
||
baud_value = _value_to_int(baud_override or self.private_cfg.baud)
|
||
if baud_value is None:
|
||
raise ValueError(f"无效波特率: {baud_override or self.private_cfg.baud}")
|
||
bootstrap_baud = 115200
|
||
use_stub = bool(self.private_cfg.flash_args.get("use_stub", True))
|
||
|
||
esp = esptool.detect_chip(
|
||
port=port,
|
||
baud=bootstrap_baud,
|
||
connect_mode=self.private_cfg.flash_args["before"],
|
||
)
|
||
try:
|
||
if use_stub:
|
||
try:
|
||
esp = cmds.run_stub(esp)
|
||
except Exception as exc: # noqa: BLE001
|
||
if on_output:
|
||
on_output(f"[警告] 启动 stub 失败，已降级为 no-stub: {exc}\n")
|
||
if baud_value != bootstrap_baud:
|
||
try:
|
||
esp.change_baud(baud_value)
|
||
except Exception as exc: # noqa: BLE001
|
||
if on_output:
|
||
on_output(
|
||
f"[警告] 切换波特率到 {baud_value} 失败，继续使用 {bootstrap_baud}: {exc}\n"
|
||
)
|
||
|
||
# Keep API flow consistent with esptool CLI: attach SPI flash before write.
|
||
cmds.attach_flash(esp, None)
|
||
|
||
for entry in entries_list:
|
||
try:
|
||
payload = entry.path.read_bytes()
|
||
except OSError as exc:
|
||
raise RuntimeError(f"固件读取失败: {entry.path.name}") from exc
|
||
try:
|
||
payload = decrypt_firmware_blob(payload, decrypt_key)
|
||
except FirmwareEnvelopeError as exc:
|
||
raise RuntimeError(f"固件解密失败: {entry.path.name}: {exc}") from exc
|
||
|
||
plain: Optional[bytearray] = None
|
||
try:
|
||
plain = bytearray(payload)
|
||
payload = b""
|
||
|
||
sha256 = _sha256_hex(plain)
|
||
report.entries.append(
|
||
FlashEntryReport(
|
||
name=entry.name,
|
||
offset=entry.offset,
|
||
size=len(plain),
|
||
sha256=sha256,
|
||
readback_sha256=None,
|
||
verified=None,
|
||
)
|
||
)
|
||
write_kwargs = {
|
||
"flash_freq": self.private_cfg.flash_args["flash_freq"],
|
||
"flash_mode": self.private_cfg.flash_args["flash_mode"],
|
||
"flash_size": self.private_cfg.flash_args["flash_size"],
|
||
"encrypt": encrypt,
|
||
}
|
||
plain_bytes = bytes(plain)
|
||
if force and entry.offset < 0x8000 and not self._is_force_bootloader_entry(entry):
|
||
raise RuntimeError(
|
||
f"force 策略拒绝低地址条目: {entry.name} @ {hex(entry.offset)}"
|
||
)
|
||
if force and self._is_force_bootloader_entry(entry):
|
||
self._guard_force_bootloader_write(
|
||
esp,
|
||
entry,
|
||
plain_bytes,
|
||
encrypt=encrypt,
|
||
)
|
||
write_kwargs["force"] = True
|
||
try:
|
||
cmds.write_flash(
|
||
esp,
|
||
[(entry.offset, plain_bytes)],
|
||
**write_kwargs,
|
||
)
|
||
except TypeError as exc:
|
||
if write_kwargs.get("force") and "force" in str(exc):
|
||
# Compatibility fallback for older esptool APIs.
|
||
write_kwargs.pop("force", None)
|
||
if on_output:
|
||
on_output("[警告] 当前 esptool API 不支持 force 参数，已回退默认写入。\n")
|
||
cmds.write_flash(
|
||
esp,
|
||
[(entry.offset, plain_bytes)],
|
||
**write_kwargs,
|
||
)
|
||
else:
|
||
raise
|
||
finally:
|
||
_wipe_bytearray(plain)
|
||
after = self.private_cfg.flash_args["after"]
|
||
if after:
|
||
after_norm = str(after).strip()
|
||
if after_norm:
|
||
after_norm = after_norm.replace("_", "-").lower()
|
||
if after_norm not in ("no-reset", "noreset", "none", "no"):
|
||
esptool.reset_chip(esp, reset_mode=after_norm)
|
||
finally:
|
||
try:
|
||
esp._port.close()
|
||
except Exception:
|
||
pass
|
||
|
||
code, output = run_esptool_api(_run, stream=True, on_output=on_output)
|
||
return code, output, report
|
||
|
||
def ensure_esptool_api_available(self) -> bool:
|
||
try:
|
||
import esptool # noqa: F401
|
||
|
||
return True
|
||
except Exception as exc: # noqa: BLE001
|
||
if self.frozen:
|
||
self._notify(
|
||
"error",
|
||
"缺少 esptool",
|
||
"内置环境缺少 esptool，请重新打包。\n\n" + str(exc),
|
||
)
|
||
return False
|
||
script_path = self.base_dir / "flash_encrypted_gui.py"
|
||
self._notify(
|
||
"error",
|
||
"缺少 esptool",
|
||
"当前 Python 环境缺少 esptool（API 模式需要同一解释器可用）。\n"
|
||
f"请使用虚拟环境运行：\n {self.python_cmd} {script_path}\n"
|
||
f"或安装：\n {sys.executable} -m pip install esptool\n\n{exc}",
|
||
)
|
||
return False
|
||
|
||
def ensure_tools_available(self) -> bool:
|
||
if not self.ensure_esptool_api_available():
|
||
return False
|
||
code, output = self.run_tool("esptool", ["version"], stream=False)
|
||
if code != 0:
|
||
if self.frozen:
|
||
self._notify("error", "缺少 esptool", "内置环境缺少 esptool，请重新打包。\n\n" + output)
|
||
return False
|
||
self._notify(
|
||
"error",
|
||
"缺少 esptool",
|
||
"未检测到 esptool。请先安装：\n"
|
||
f"{self.python_cmd} -m pip install esptool\n\n{output}",
|
||
)
|
||
return False
|
||
|
||
if self.frozen:
|
||
try:
|
||
import importlib
|
||
|
||
importlib.import_module("espefuse")
|
||
except Exception as exc: # noqa: BLE001
|
||
self._notify(
|
||
"error",
|
||
"缺少 espefuse",
|
||
"内置环境缺少 espefuse，请重新打包。\n\n" + str(exc),
|
||
)
|
||
return False
|
||
return True
|
||
|
||
code, output = self.run_tool("espefuse", ["--help"], stream=False)
|
||
if code != 0:
|
||
self._notify(
|
||
"error",
|
||
"缺少 espefuse",
|
||
"未检测到 espefuse。请先安装：\n"
|
||
f"{self.python_cmd} -m pip install esptool\n\n{output}",
|
||
)
|
||
return False
|
||
return True
|
||
|
||
def run_tool(
|
||
self,
|
||
tool: str,
|
||
args: Iterable[str],
|
||
stream: bool,
|
||
on_output: Optional[Callable[[str], None]] = None,
|
||
) -> Tuple[int, str]:
|
||
if self.frozen:
|
||
return run_tool_module(tool, args, cwd=self.base_dir, stream=stream, on_output=on_output)
|
||
if self.python_cmd is None:
|
||
return 1, "Python 环境未就绪。"
|
||
cmd = [str(self.python_cmd), "-m", tool, *args]
|
||
return run_command(cmd, cwd=self.base_dir, stream=stream, on_output=on_output)
|
||
|
||
def do_flash(
|
||
self,
|
||
port: str,
|
||
encrypt: bool,
|
||
entries: Iterable[FlashSource],
|
||
on_output: Optional[Callable[[str], None]],
|
||
baud_override: Optional[str] = None,
|
||
force: bool = False,
|
||
) -> Tuple[int, str, FlashReport]:
|
||
return self.do_write_flash(
|
||
port,
|
||
entries,
|
||
encrypt=encrypt,
|
||
on_output=on_output,
|
||
baud_override=baud_override,
|
||
force=force,
|
||
)
|
||
|
||
def run(self) -> None:
|
||
self.root.mainloop()
|
||
|
||
|
||
def main() -> None:
|
||
try:
|
||
app = FlashGUI()
|
||
except (FileNotFoundError, PermissionError, ValueError, KeyError, TypeError) as exc:
|
||
try:
|
||
messagebox.showerror("配置错误", str(exc))
|
||
except tk.TclError:
|
||
print(exc, file=sys.stderr)
|
||
raise SystemExit(1) from exc
|
||
app.run()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
main()
|
||
except KeyboardInterrupt:
|
||
sys.exit(1)
|