"""NAND transfer GUI: pick COM port, local folder, FLASH folder, then batch upload.""" import queue import threading import tkinter as tk from pathlib import Path from tkinter import filedialog, messagebox, simpledialog, ttk import nand_put as nx try: from serial.tools import list_ports except ImportError: list_ports = None LOCAL_DEFAULT = Path(r"E:\PET\EAF") FLASH_DEFAULT = "/extflash/eaf" def com_choices(): if list_ports is None: return [] items = [] for port in list_ports.comports(): extra = port.description or "" items.append((port.device, f"{port.device} {extra}".rstrip())) items.sort(key=lambda item: (0 if "303A" in item[1].upper() or "SERIAL/JTAG" in item[1].upper() else 1, item[0])) return items def join_flash(root, *parts): path = root.rstrip("/") for part in parts: if part: path += "/" + str(part) return path class App(tk.Tk): def __init__(self): super().__init__() self.title("NAND 传输") self.geometry("820x560") self.minsize(640, 420) LOCAL_DEFAULT.mkdir(parents=True, exist_ok=True) self._jobs = queue.Queue() self._stop = threading.Event() self._busy = False self._build() self.refresh_ports() self.after(80, self._pump) def _build(self): pad = {"padx": 8, "pady": 4} top = ttk.Frame(self) top.pack(fill="x", **pad) ttk.Label(top, text="端口").grid(row=0, column=0, sticky="w") self.port = ttk.Combobox(top, width=42, state="readonly") self.port.grid(row=0, column=1, sticky="ew", padx=4) ttk.Button(top, text="刷新", command=self.refresh_ports).grid(row=0, column=2) ttk.Label(top, text="本地目录").grid(row=1, column=0, sticky="w") self.local = ttk.Entry(top) self.local.insert(0, str(LOCAL_DEFAULT)) self.local.grid(row=1, column=1, sticky="ew", padx=4) ttk.Button(top, text="浏览", command=self.browse_local).grid(row=1, column=2) ttk.Label(top, text="FLASH目录").grid(row=2, column=0, sticky="w") self.flash = ttk.Entry(top) self.flash.insert(0, FLASH_DEFAULT) self.flash.grid(row=2, column=1, sticky="ew", padx=4) top.columnconfigure(1, weight=1) bar = ttk.Frame(self) bar.pack(fill="x", **pad) self.btn_info = ttk.Button(bar, text="查询", command=self.on_info) self.btn_info.pack(side="left", padx=(0, 6)) self.btn_list = ttk.Button(bar, text="列目录", command=self.on_list) self.btn_list.pack(side="left", padx=(0, 6)) self.btn_put = ttk.Button(bar, text="批量写入", command=self.on_upload) self.btn_put.pack(side="left", padx=(0, 6)) self.btn_format = ttk.Button(bar, text="格式化", command=self.on_format) self.btn_format.pack(side="left", padx=(0, 6)) self.btn_stop = ttk.Button(bar, text="停止", command=self.on_stop, state="disabled") self.btn_stop.pack(side="left") self.progress = ttk.Progressbar(self, mode="determinate") self.progress.pack(fill="x", padx=8, pady=(0, 4)) self.status = ttk.Label(self, text="空闲") self.status.pack(anchor="w", padx=8) pane = ttk.Panedwindow(self, orient="vertical") pane.pack(fill="both", expand=True, padx=8, pady=4) list_frame = ttk.Labelframe(pane, text="FLASH 目录") log_frame = ttk.Labelframe(pane, text="日志") pane.add(list_frame, weight=1) pane.add(log_frame, weight=1) self.listing = tk.Listbox(list_frame, font=("Consolas", 10), height=8) scroll_list = ttk.Scrollbar(list_frame, command=self.listing.yview) self.listing.configure(yscrollcommand=scroll_list.set) self.listing.pack(side="left", fill="both", expand=True) scroll_list.pack(side="right", fill="y") self.log = tk.Text(log_frame, height=10, wrap="word", state="disabled") scroll_log = ttk.Scrollbar(log_frame, command=self.log.yview) self.log.configure(yscrollcommand=scroll_log.set) self.log.pack(side="left", fill="both", expand=True) scroll_log.pack(side="right", fill="y") def refresh_ports(self): choices = com_choices() labels = [label for _, label in choices] current = self.port.get() self.port["values"] = labels if not labels: self.port.set("") self._append("没有串口。确认板子已接上，并关掉其它占用 COM 的程序。") return devices = {label: device for device, label in choices} self._port_map = devices if current in devices: self.port.set(current) else: self.port.current(0) def browse_local(self): start = self.local.get().strip() or str(LOCAL_DEFAULT) picked = filedialog.askdirectory(title="选择本地目录", initialdir=start) if picked: self.local.delete(0, "end") self.local.insert(0, picked) def selected_port(self): label = self.port.get().strip() if not label: raise ValueError("请选择串口") return getattr(self, "_port_map", {}).get(label, label.split()[0]) def flash_dir(self): path = self.flash.get().strip().replace("\\", "/") if path != nx.ROOT and path.endswith("/"): path = path.rstrip("/") nx.path_bytes(path, allow_root=True) return path def on_info(self): self._start("查询", self._job_info) def on_list(self): self._start("列目录", self._job_list) def on_upload(self): self._start("批量写入", self._job_upload) def on_format(self): if not messagebox.askokcancel( "格式化 NAND", "将清空外置 NAND 上的全部文件。\n新盘第一次使用需要格式化；已有表情文件的盘不要点。"): return typed = simpledialog.askstring("确认格式化", "请输入 FORMAT /extflash") if typed != "FORMAT /extflash": self._append("格式化已取消") return self._start("格式化", self._job_format) def on_stop(self): self._stop.set() self._append("停止：当前文件结束后不再继续") def _start(self, title, job): if self._busy: return try: port = self.selected_port() flash = self.flash_dir() local = Path(self.local.get().strip() or str(LOCAL_DEFAULT)) except Exception as exc: messagebox.showerror("参数错误", str(exc)) return self._busy = True self._stop.clear() self.btn_info.configure(state="disabled") self.btn_list.configure(state="disabled") self.btn_put.configure(state="disabled") self.btn_format.configure(state="disabled") self.btn_stop.configure(state="normal") self.status.configure(text=title + "…") threading.Thread(target=self._run, args=(title, job, port, local, flash), daemon=True).start() def _run(self, title, job, port, local, flash): try: if nx.serial is None: raise ValueError("pyserial missing: py -3 -m pip install pyserial") with nx.open_port(port) as serial_port: client = nx.Client(serial_port) client.connect() job(client, local, flash) self._post(lambda: self.status.configure(text=title + "完成")) except Exception as exc: self._append("错误: " + str(exc)) self._post(lambda: self.status.configure(text=title + "失败")) finally: self._post(self._idle) def _job_info(self, client, local, flash): page, pages, mounted, total, free = client.info() self._append( f"NAND {page * pages} bytes; mounted={mounted}; FAT={total} bytes, free={free}") if not mounted: self._append("未挂载：新盘需要先点「格式化」，再列目录或写入。") def _job_format(self, client, local, flash): self._append("正在格式化 NAND，可能需要一两分钟…") nx.format_volume(client) self._append("格式化完成并已重新挂载") self._job_info(client, local, flash) def _job_list(self, client, local, flash): rows = [] for is_dir, name, size in client.listdir(flash): if is_dir: rows.append(f"DIR {name}") else: rows.append(f"FILE {name} {size}") if not rows: rows.append("(空目录)") self._append(f"列目录 {flash} ({len(rows)} 项)") self._post(lambda: self._show_listing(flash, rows)) def _job_upload(self, client, local, flash): if not local.is_dir(): raise ValueError(f"本地目录不存在: {local}") entries = list(nx.directory_entries(local)) files = [(path, parts) for path, parts, is_dir in entries if not is_dir] dirs = [parts for _, parts, is_dir in entries if is_dir] if not files and not dirs: self._append(f"本地目录没有可上传的文件: {local}") return total = len(files) self._post(lambda: self._set_progress(0, max(total, 1))) if flash != nx.ROOT: client.call(nx.MKDIR, nx.path_bytes(flash)) for parts in dirs: if self._stop.is_set(): self._append("已停止") return client.call(nx.MKDIR, nx.path_bytes(join_flash(flash, *parts))) done = 0 failed = 0 for path, parts in files: if self._stop.is_set(): self._append("已停止") break dest = join_flash(flash, *parts) self._append(f"写入 {path.name} -> {dest}") try: client.upload(path, dest, log=self._append) done += 1 except Exception as exc: failed += 1 self._append(f"失败 {path.name}: {exc}") n = done + failed self._post(lambda n=n, total=total: self._set_progress(n, total)) self._append(f"批量结束：成功 {done}/{total}" + (f"，失败 {failed}" if failed else "")) try: self._job_list(client, local, flash) except Exception as exc: self._append("写入后列目录失败: " + str(exc)) def _set_progress(self, value, maximum): self.progress.configure(maximum=max(maximum, 1), value=value) self.status.configure(text=f"{value}/{maximum}") def _show_listing(self, flash, rows): self.listing.delete(0, "end") self.listing.insert("end", flash) for row in rows: self.listing.insert("end", " " + row) def _idle(self): self._busy = False self.btn_info.configure(state="normal") self.btn_list.configure(state="normal") self.btn_put.configure(state="normal") self.btn_format.configure(state="normal") self.btn_stop.configure(state="disabled") def _append(self, line): self._jobs.put(line) def _post(self, fn): self._jobs.put(fn) def _pump(self): try: while True: item = self._jobs.get_nowait() if callable(item): item() else: self.log.configure(state="normal") self.log.insert("end", item + "\n") self.log.see("end") self.log.configure(state="disabled") except queue.Empty: pass self.after(80, self._pump) def main(): try: from ctypes import windll windll.shcore.SetProcessDpiAwareness(1) except Exception: pass try: App().mainloop() except Exception as exc: try: messagebox.showerror("NAND 传输", str(exc)) except Exception: raise if __name__ == "__main__": main()