From 6cc3e55ebba71de268bcd54ca55b97a5c2a8d808 Mon Sep 17 00:00:00 2001 From: "xinzhu.yin" Date: Tue, 21 Apr 2026 14:06:48 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0AI=E5=9B=BE=E7=89=87=E5=8A=9F?= =?UTF-8?q?=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/services/__init__.py | 0 app/services/ai_image.py | 224 +++++++++++++++++++++++ app/views/panels/ai_image_panel.py | 285 +++++++++++++++++++++++++++++ app/views/panels/main_layout.py | 14 +- pqAutomationApp.py | 10 +- settings/pq_config.json | 2 +- 6 files changed, 532 insertions(+), 3 deletions(-) create mode 100644 app/services/__init__.py create mode 100644 app/services/ai_image.py create mode 100644 app/views/panels/ai_image_panel.py diff --git a/app/services/__init__.py b/app/services/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/services/ai_image.py b/app/services/ai_image.py new file mode 100644 index 0000000..0dedddf --- /dev/null +++ b/app/services/ai_image.py @@ -0,0 +1,224 @@ +"""AI 图片生成服务：后端请求 + 本地缓存管理。 + +API 端点待接入，当前通过 ``set_api_caller`` 注入具体实现。 +缓存目录：``settings/ai_image_cache/``，每张图片有同名的 ``.json`` 侧车记录。 +""" + +from __future__ import annotations + +import datetime as _dt +import hashlib +import json +import os +import shutil +import threading +from dataclasses import dataclass, asdict +from typing import Callable, List, Optional + + +# ---------- 常量 ---------- + +_CACHE_DIRNAME = os.path.join("settings", "ai_image_cache") +_META_SUFFIX = ".json" +_SUPPORTED_IMG_EXT = (".png", ".jpg", ".jpeg", ".bmp", ".webp") + + +# ---------- 数据结构 ---------- + + +@dataclass +class AIImageRecord: + """一条缓存记录。""" + + id: str + prompt: str + image_path: str + created_at: str # ISO8601 + extra: Optional[dict] = None + + def to_json(self) -> str: + return json.dumps(asdict(self), ensure_ascii=False, indent=2) + + +# ---------- API 注入 ---------- + +# 调用签名: ``fn(prompt: str) -> (image_bytes: bytes, image_ext: str, extra: dict|None)`` +# ``image_ext`` 例如 ``".png"``；``extra`` 可为 None。 +_ApiCaller = Callable[[str], tuple] + +_api_caller: Optional[_ApiCaller] = None + + +def set_api_caller(fn: Optional[_ApiCaller]) -> None: + """注入真实的后端 API 调用函数。在 API 就绪前可保持为 None。""" + global _api_caller + _api_caller = fn + + +def has_api() -> bool: + return _api_caller is not None + + +# ---------- 缓存路径工具 ---------- + + +def get_cache_dir(base_dir: Optional[str] = None) -> str: + """返回缓存目录，如不存在则创建。``base_dir`` 默认使用当前工作目录。""" + root = base_dir if base_dir else os.getcwd() + path = os.path.join(root, _CACHE_DIRNAME) + os.makedirs(path, exist_ok=True) + return path + + +def _make_id(prompt: str) -> str: + stamp = _dt.datetime.now().strftime("%Y%m%d_%H%M%S_%f") + digest = hashlib.md5(prompt.encode("utf-8")).hexdigest()[:8] + return f"{stamp}_{digest}" + + +def _meta_path_for(image_path: str) -> str: + return os.path.splitext(image_path)[0] + _META_SUFFIX + + +# ---------- 读写 ---------- + + +def list_records(base_dir: Optional[str] = None) -> List[AIImageRecord]: + """列出缓存目录下的所有记录，按创建时间倒序（最新在前）。""" + cache_dir = get_cache_dir(base_dir) + records: List[AIImageRecord] = [] + for name in os.listdir(cache_dir): + full = os.path.join(cache_dir, name) + if not (os.path.isfile(full) and name.lower().endswith(_SUPPORTED_IMG_EXT)): + continue + meta_path = _meta_path_for(full) + prompt = "" + created_at = "" + extra = None + rec_id = os.path.splitext(name)[0] + if os.path.isfile(meta_path): + try: + with open(meta_path, "r", encoding="utf-8") as f: + data = json.load(f) + prompt = data.get("prompt", "") + created_at = data.get("created_at", "") + extra = data.get("extra") + rec_id = data.get("id", rec_id) + except Exception: + pass + if not created_at: + # fallback 到文件 mtime + try: + mtime = os.path.getmtime(full) + created_at = _dt.datetime.fromtimestamp(mtime).isoformat() + except Exception: + created_at = "" + records.append( + AIImageRecord( + id=rec_id, + prompt=prompt, + image_path=full, + created_at=created_at, + extra=extra, + ) + ) + records.sort(key=lambda r: r.created_at, reverse=True) + return records + + +def save_image_to_cache( + prompt: str, + image_bytes: bytes, + image_ext: str = ".png", + extra: Optional[dict] = None, + base_dir: Optional[str] = None, +) -> AIImageRecord: + """把生成的图片字节写入缓存，返回记录。""" + if not image_ext.startswith("."): + image_ext = "." + image_ext + if image_ext.lower() not in _SUPPORTED_IMG_EXT: + image_ext = ".png" + cache_dir = get_cache_dir(base_dir) + rec_id = _make_id(prompt) + image_path = os.path.join(cache_dir, f"{rec_id}{image_ext}") + with open(image_path, "wb") as f: + f.write(image_bytes) + + record = AIImageRecord( + id=rec_id, + prompt=prompt, + image_path=image_path, + created_at=_dt.datetime.now().isoformat(timespec="seconds"), + extra=extra, + ) + try: + with open(_meta_path_for(image_path), "w", encoding="utf-8") as f: + f.write(record.to_json()) + except Exception: + pass + return record + + +def delete_record(record: AIImageRecord) -> bool: + """删除一条缓存记录（图片 + 侧车）。返回是否成功。""" + ok = True + for p in (record.image_path, _meta_path_for(record.image_path)): + try: + if os.path.isfile(p): + os.remove(p) + except Exception: + ok = False + return ok + + +def export_record(record: AIImageRecord, dest_path: str) -> None: + """把缓存中的图片另存到 ``dest_path``。""" + shutil.copyfile(record.image_path, dest_path) + + +# ---------- 异步请求 ---------- + + +def request_image_async( + prompt: str, + on_success: Callable[[AIImageRecord], None], + on_error: Callable[[Exception], None], + base_dir: Optional[str] = None, +) -> threading.Thread: + """在后台线程请求 API → 写入缓存 → 回调。 + + ``on_success`` / ``on_error`` 会在 **工作线程** 中被调用；UI 侧若需 + 切回主线程，请在回调内部自行用 ``root.after(0, ...)``。 + """ + + def _worker(): + try: + if _api_caller is None: + raise RuntimeError("AI 图片 API 尚未接入，请调用 set_api_caller 注入") + image_bytes, image_ext, extra = _normalize_api_result(_api_caller(prompt)) + record = save_image_to_cache( + prompt=prompt, + image_bytes=image_bytes, + image_ext=image_ext, + extra=extra, + base_dir=base_dir, + ) + on_success(record) + except Exception as exc: + on_error(exc) + + t = threading.Thread(target=_worker, daemon=True) + t.start() + return t + + +def _normalize_api_result(result): + """允许 API 返回 ``bytes`` 或 ``(bytes, ext)`` 或 ``(bytes, ext, extra)``。""" + if isinstance(result, (bytes, bytearray)): + return bytes(result), ".png", None + if isinstance(result, tuple): + if len(result) == 2: + return bytes(result[0]), str(result[1]), None + if len(result) == 3: + return bytes(result[0]), str(result[1]), result[2] + raise ValueError("API 返回格式不支持，需为 bytes 或 (bytes, ext[, extra])") diff --git a/app/views/panels/ai_image_panel.py b/app/views/panels/ai_image_panel.py new file mode 100644 index 0000000..a197595 --- /dev/null +++ b/app/views/panels/ai_image_panel.py @@ -0,0 +1,285 @@ +"""AI 图片对话面板：输入提示 → 生成 → 显示 → 列表切换/保存/删除。""" + +from __future__ import annotations + +import os +import tkinter as tk +from tkinter import filedialog, messagebox +from typing import List, Optional + +import ttkbootstrap as ttk +from PIL import Image, ImageTk + +from app.services import ai_image as _svc + + +# ---------------- 面板创建 ---------------- + + +def create_ai_image_panel(self): + """创建 AI 图片对话面板，并注册到面板管理。""" + frame = ttk.Frame(self.content_frame) + self.ai_image_frame = frame + + # 内部状态 + self.ai_image_records: List[_svc.AIImageRecord] = [] + self.ai_image_current: Optional[_svc.AIImageRecord] = None + self.ai_image_photo = None # 防止 ImageTk.PhotoImage 被 GC + self._ai_image_requesting = False + + container = ttk.Frame(frame, padding=10) + container.pack(fill=tk.BOTH, expand=True) + + # 左列：图片列表 + left = ttk.Frame(container) + left.pack(side=tk.LEFT, fill=tk.Y, padx=(0, 10)) + + ttk.Label(left, text="历史图片", font=("微软雅黑", 10, "bold")).pack( + anchor=tk.W, pady=(0, 4) + ) + + list_wrap = ttk.Frame(left) + list_wrap.pack(fill=tk.Y, expand=False) + + scroll = ttk.Scrollbar(list_wrap, orient=tk.VERTICAL) + self.ai_image_listbox = tk.Listbox( + list_wrap, + width=28, + height=22, + activestyle="dotbox", + yscrollcommand=scroll.set, + ) + scroll.config(command=self.ai_image_listbox.yview) + self.ai_image_listbox.pack(side=tk.LEFT, fill=tk.Y) + scroll.pack(side=tk.RIGHT, fill=tk.Y) + self.ai_image_listbox.bind("<>", lambda e: _on_list_select(self)) + + btn_row = ttk.Frame(left) + btn_row.pack(fill=tk.X, pady=(6, 0)) + ttk.Button( + btn_row, text="保存", bootstyle="success-outline", width=8, + command=lambda: _save_current(self), + ).pack(side=tk.LEFT, padx=(0, 4)) + ttk.Button( + btn_row, text="删除", bootstyle="danger-outline", width=8, + command=lambda: _delete_current(self), + ).pack(side=tk.LEFT, padx=(0, 4)) + ttk.Button( + btn_row, text="刷新", bootstyle="secondary-outline", width=8, + command=lambda: reload_ai_image_list(self), + ).pack(side=tk.LEFT) + + # 右列：预览 + 输入 + right = ttk.Frame(container) + right.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) + + preview_frame = ttk.LabelFrame(right, text="图片预览", padding=6) + preview_frame.pack(fill=tk.BOTH, expand=True) + + self.ai_image_canvas = tk.Canvas( + preview_frame, bg="#1e1e1e", highlightthickness=0 + ) + self.ai_image_canvas.pack(fill=tk.BOTH, expand=True) + self.ai_image_canvas.bind("", lambda e: _redraw_preview(self)) + + meta_row = ttk.Frame(right) + meta_row.pack(fill=tk.X, pady=(4, 4)) + self.ai_image_meta_var = tk.StringVar(value="未选择图片") + ttk.Label( + meta_row, textvariable=self.ai_image_meta_var, + foreground="#666", font=("微软雅黑", 9), + ).pack(side=tk.LEFT) + + # 输入区 + input_frame = ttk.LabelFrame(right, text="提示输入（Ctrl+Enter 发送）", padding=6) + input_frame.pack(fill=tk.X, pady=(4, 0)) + + self.ai_image_input = tk.Text(input_frame, height=3, wrap=tk.WORD) + self.ai_image_input.pack(fill=tk.X, side=tk.TOP) + self.ai_image_input.bind("", lambda e: (_send_prompt(self), "break")) + + send_row = ttk.Frame(input_frame) + send_row.pack(fill=tk.X, pady=(4, 0)) + self.ai_image_status_var = tk.StringVar(value="就绪") + ttk.Label( + send_row, textvariable=self.ai_image_status_var, + foreground="#888", font=("微软雅黑", 9), + ).pack(side=tk.LEFT) + self.ai_image_send_btn = ttk.Button( + send_row, text="发送", bootstyle="primary", width=10, + command=lambda: _send_prompt(self), + ) + self.ai_image_send_btn.pack(side=tk.RIGHT) + + # 注册面板 + self.register_panel("ai_image", frame, None, "ai_image_visible") + self.ai_image_visible = False + + # 初次加载缓存 + reload_ai_image_list(self) + + +def toggle_ai_image_panel(self): + """切换 AI 图片面板显隐。""" + self.show_panel("ai_image") + + +# ---------------- 列表 / 选中 ---------------- + + +def reload_ai_image_list(self): + """重新扫描缓存并刷新列表。""" + self.ai_image_records = _svc.list_records() + self.ai_image_listbox.delete(0, tk.END) + for rec in self.ai_image_records: + label = _format_list_label(rec) + self.ai_image_listbox.insert(tk.END, label) + if self.ai_image_records: + self.ai_image_listbox.selection_clear(0, tk.END) + self.ai_image_listbox.selection_set(0) + self.ai_image_listbox.activate(0) + _select_record(self, self.ai_image_records[0]) + else: + self.ai_image_current = None + self.ai_image_photo = None + self.ai_image_canvas.delete("all") + self.ai_image_meta_var.set("暂无缓存图片") + + +def _format_list_label(rec: _svc.AIImageRecord) -> str: + when = (rec.created_at or "")[:19].replace("T", " ") + preview = (rec.prompt or "(无提示)").strip().splitlines()[0] + if len(preview) > 18: + preview = preview[:18] + "…" + return f"{when} {preview}" if when else preview + + +def _on_list_select(self): + sel = self.ai_image_listbox.curselection() + if not sel: + return + idx = sel[0] + if 0 <= idx < len(self.ai_image_records): + _select_record(self, self.ai_image_records[idx]) + + +def _select_record(self, rec: _svc.AIImageRecord): + self.ai_image_current = rec + self.ai_image_meta_var.set( + f"{os.path.basename(rec.image_path)} | {rec.created_at}" + ) + _redraw_preview(self) + + +# ---------------- 预览绘制 ---------------- + + +def _redraw_preview(self): + rec = getattr(self, "ai_image_current", None) + canvas = self.ai_image_canvas + canvas.delete("all") + if rec is None or not os.path.isfile(rec.image_path): + return + cw = canvas.winfo_width() or 1 + ch = canvas.winfo_height() or 1 + try: + img = Image.open(rec.image_path) + except Exception as exc: + canvas.create_text(cw // 2, ch // 2, text=f"加载失败: {exc}", fill="#f66") + return + iw, ih = img.size + scale = min(cw / iw, ch / ih, 1.0) + nw, nh = max(1, int(iw * scale)), max(1, int(ih * scale)) + img_resized = img.resize((nw, nh), Image.LANCZOS) + self.ai_image_photo = ImageTk.PhotoImage(img_resized) + canvas.create_image(cw // 2, ch // 2, image=self.ai_image_photo, anchor="center") + + +# ---------------- 发送 / 保存 / 删除 ---------------- + + +def _send_prompt(self): + if getattr(self, "_ai_image_requesting", False): + return + prompt = self.ai_image_input.get("1.0", tk.END).strip() + if not prompt: + messagebox.showinfo("提示", "请输入内容") + return + if not _svc.has_api(): + messagebox.showerror( + "API 未配置", + "AI 图片 API 尚未接入。\n请在启动时通过 " + "app.services.ai_image.set_api_caller(...) 注入真实实现。", + ) + return + + _set_requesting(self, True) + self.ai_image_status_var.set("请求中…") + + def _success(record): + self.root.after(0, lambda: _on_request_done(self, record, None)) + + def _error(exc): + self.root.after(0, lambda: _on_request_done(self, None, exc)) + + _svc.request_image_async(prompt, on_success=_success, on_error=_error) + + +def _set_requesting(self, flag: bool): + self._ai_image_requesting = flag + try: + self.ai_image_send_btn.configure(state=tk.DISABLED if flag else tk.NORMAL) + except Exception: + pass + + +def _on_request_done(self, record: Optional[_svc.AIImageRecord], exc: Optional[Exception]): + _set_requesting(self, False) + if exc is not None: + self.ai_image_status_var.set(f"失败: {exc}") + messagebox.showerror("生成失败", str(exc)) + return + self.ai_image_status_var.set("完成") + self.ai_image_input.delete("1.0", tk.END) + reload_ai_image_list(self) + # 定位到新生成项（最新在前） + if record is not None and self.ai_image_records: + for i, r in enumerate(self.ai_image_records): + if r.id == record.id: + self.ai_image_listbox.selection_clear(0, tk.END) + self.ai_image_listbox.selection_set(i) + self.ai_image_listbox.activate(i) + _select_record(self, r) + break + + +def _save_current(self): + rec = getattr(self, "ai_image_current", None) + if rec is None: + messagebox.showinfo("提示", "请先选择一张图片") + return + ext = os.path.splitext(rec.image_path)[1] or ".png" + dest = filedialog.asksaveasfilename( + title="另存为", + defaultextension=ext, + initialfile=os.path.basename(rec.image_path), + filetypes=[("图片", "*.png;*.jpg;*.jpeg;*.bmp;*.webp"), ("所有文件", "*.*")], + ) + if not dest: + return + try: + _svc.export_record(rec, dest) + messagebox.showinfo("成功", f"已保存到:\n{dest}") + except Exception as exc: + messagebox.showerror("保存失败", str(exc)) + + +def _delete_current(self): + rec = getattr(self, "ai_image_current", None) + if rec is None: + messagebox.showinfo("提示", "请先选择一张图片") + return + if not messagebox.askyesno("确认删除", f"确定删除这张缓存图片吗？\n{os.path.basename(rec.image_path)}"): + return + _svc.delete_record(rec) + reload_ai_image_list(self) diff --git a/app/views/panels/main_layout.py b/app/views/panels/main_layout.py index 0ec0fab..cc7a388 100644 --- a/app/views/panels/main_layout.py +++ b/app/views/panels/main_layout.py @@ -425,12 +425,24 @@ def create_test_type_frame(self): ) self.local_dimming_btn.pack(fill=tk.X, padx=0, pady=1) - # 注册面板按钮（只保留日志） + # AI 图片对话按钮 + self.ai_image_btn = ttk.Button( + self.sidebar_frame, + text="AI 图片", + style="Sidebar.TButton", + command=self.toggle_ai_image_panel, + takefocus=False, + ) + self.ai_image_btn.pack(fill=tk.X, padx=0, pady=1) + + # 注册面板按钮 if hasattr(self, "panels"): if "log" in self.panels: self.panels["log"]["button"] = self.log_btn if "local_dimming" in self.panels: self.panels["local_dimming"]["button"] = self.local_dimming_btn + if "ai_image" in self.panels: + self.panels["ai_image"]["button"] = self.ai_image_btn def update_config_info_display(self): diff --git a/pqAutomationApp.py b/pqAutomationApp.py index 16b4a29..eaa5a30 100644 --- a/pqAutomationApp.py +++ b/pqAutomationApp.py @@ -22,6 +22,7 @@ from app.views.panels import custom_template_panel as _ctp from app.views.panels import side_panels as _sp from app.views.panels import cct_panel as _ccp from app.views.panels import main_layout as _main +from app.views.panels import ai_image_panel as _aip from app.views import panel_manager as PM # Step 0/1 重构：资源工具和纯算法已迁移到 app/ 包，这里重新导入以保持 @@ -196,6 +197,8 @@ class PQAutomationApp: self.create_log_panel() # 创建 Local Dimming 面板 self.create_local_dimming_panel() + # 创建 AI 图片对话面板 + self.create_ai_image_panel() # 创建测试类型选择区域 self.create_test_type_frame() # 创建操作按钮区域 @@ -323,6 +326,11 @@ class PQAutomationApp: toggle_log_panel = _sp.toggle_log_panel update_sidebar_selection = _sp.update_sidebar_selection + # ---- AI 图片对话面板 ---- + create_ai_image_panel = _aip.create_ai_image_panel + toggle_ai_image_panel = _aip.toggle_ai_image_panel + reload_ai_image_list = _aip.reload_ai_image_list + # ---- 单步调试面板（统一实现，委托到 side_panels 模块） ---- _toggle_debug_panel = _sp._toggle_debug_panel toggle_screen_debug_panel = _sp.toggle_screen_debug_panel @@ -517,7 +525,7 @@ class PQAutomationApp: def change_test_type(self, test_type): """切换测试类型""" # 切换测试类型时，自动隐藏日志面板和 Local Dimming 面板 - if self.current_panel in ("log", "local_dimming"): + if self.current_panel in ("log", "local_dimming", "ai_image"): self.hide_all_panels() self._save_cct_params_before_test_type_switch() self._apply_current_test_type(test_type) diff --git a/settings/pq_config.json b/settings/pq_config.json index 02b8611..d7b733a 100644 --- a/settings/pq_config.json +++ b/settings/pq_config.json @@ -1,5 +1,5 @@ { - "current_test_type": "sdr_movie", + "current_test_type": "screen_module", "test_types": { "screen_module": { "name": "屏模组性能测试",