添加AI图片功能

This commit is contained in:
xinzhu.yin
2026-04-21 14:06:48 +08:00
parent 6b8bfe06b9
commit 6cc3e55ebb
6 changed files with 532 additions and 3 deletions

0
app/services/__init__.py Normal file
View File

224
app/services/ai_image.py Normal file
View File

@@ -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])")

View File

@@ -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("<<ListboxSelect>>", 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("<Configure>", 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("<Control-Return>", 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)

View File

@@ -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):

View File

@@ -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)

View File

@@ -1,5 +1,5 @@
{
"current_test_type": "sdr_movie",
"current_test_type": "screen_module",
"test_types": {
"screen_module": {
"name": "屏模组性能测试",