562 lines
19 KiB
Python
562 lines
19 KiB
Python
"""AI 图片对话面板：输入提示 → 生成 → 显示 → 列表切换/保存/删除。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import threading
|
||
import tkinter as tk
|
||
from tkinter import filedialog, messagebox, simpledialog
|
||
|
||
|
||
import ttkbootstrap as ttk
|
||
from PIL import Image, ImageTk
|
||
|
||
from app.services import ai_image as _svc
|
||
from drivers.ucd_helpers import get_current_resolution, send_image_pattern
|
||
|
||
|
||
# ---------------- 面板创建 ----------------
|
||
|
||
|
||
def create_ai_image_panel(self):
|
||
"""创建 AI 图片对话面板，并注册到面板管理。"""
|
||
frame = ttk.Frame(self.content_frame)
|
||
self.ai_image_frame = frame
|
||
|
||
# 内部状态
|
||
self.ai_image_records = [] # list[AIImageRecord]
|
||
self.ai_image_current = None # 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)
|
||
|
||
# 左列：图片列表
|
||
# 使用 grid + 权重，让右侧预览区优先占据剩余空间。
|
||
container.columnconfigure(0, weight=0)
|
||
container.columnconfigure(1, weight=1)
|
||
container.rowconfigure(0, weight=1)
|
||
|
||
left = ttk.Frame(container, width=360)
|
||
left.grid(row=0, column=0, sticky=tk.NS, padx=(0, 10))
|
||
left.grid_propagate(False)
|
||
|
||
ttk.Label(left, text="历史图片", font=("微软雅黑", 10, "bold")).pack(
|
||
anchor=tk.W, pady=(0, 4)
|
||
)
|
||
|
||
list_wrap = ttk.Frame(left, padding=2)
|
||
list_wrap.pack(fill=tk.BOTH, expand=True)
|
||
|
||
scroll = ttk.Scrollbar(list_wrap, orient=tk.VERTICAL)
|
||
self.ai_image_listbox = tk.Listbox(
|
||
list_wrap,
|
||
width=34,
|
||
height=1, # 由 pack fill/expand 撑满，height 仅为最小保底
|
||
activestyle="none",
|
||
font=("微软雅黑", 9),
|
||
bd=1,
|
||
relief=tk.FLAT,
|
||
highlightthickness=1,
|
||
highlightbackground="#d8d8d8",
|
||
highlightcolor="#4a90e2",
|
||
selectbackground="#2b6cb0",
|
||
selectforeground="#ffffff",
|
||
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))
|
||
# 右键菜单：发送到 UCD / 重命名 / 另存为 / 删除
|
||
# 索引: 0=发送, 1=sep, 2=重命名, 3=另存为, 4=删除
|
||
self.ai_image_menu = tk.Menu(self.root, tearoff=0)
|
||
self.ai_image_menu.add_command(
|
||
label="发送图片",
|
||
command=lambda: _send_to_ucd(self),
|
||
)
|
||
self.ai_image_menu.add_separator()
|
||
self.ai_image_menu.add_command(
|
||
label="重命名…",
|
||
command=lambda: _rename_current(self),
|
||
)
|
||
self.ai_image_menu.add_command(
|
||
label="另存为…",
|
||
command=lambda: _save_current(self),
|
||
)
|
||
self.ai_image_menu.add_command(
|
||
label="删除",
|
||
command=lambda: _delete_current(self),
|
||
)
|
||
self.ai_image_listbox.bind(
|
||
"<Button-3>",
|
||
lambda e: _show_list_context_menu(self, e),
|
||
)
|
||
|
||
btn_row = ttk.Frame(left)
|
||
btn_row.pack(fill=tk.X, pady=(6, 0))
|
||
self.ai_image_send_ucd_btn = ttk.Button(
|
||
btn_row, text="发送", bootstyle="info-outline", width=8,
|
||
command=lambda: _send_to_ucd(self),
|
||
)
|
||
self.ai_image_send_ucd_btn.pack(side=tk.LEFT, padx=(0, 4))
|
||
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.grid(row=0, column=1, sticky=tk.NSEW)
|
||
|
||
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:
|
||
# 分辨率前缀：优先从 extra["size"] 取，其次从文件名猜
|
||
size_tag = ""
|
||
extra = rec.extra or {}
|
||
if isinstance(extra, dict) and extra.get("size"):
|
||
size_tag = f"[{extra['size']}] "
|
||
else:
|
||
try:
|
||
from PIL import Image as _Im
|
||
with _Im.open(rec.image_path) as _im:
|
||
size_tag = f"[{_im.width}×{_im.height}] "
|
||
except Exception:
|
||
pass
|
||
|
||
prompt_line = (rec.prompt or "(无提示)").strip().splitlines()[0]
|
||
# 剩余可用宽度（width=34）去掉 size_tag
|
||
max_prompt = 34 - len(size_tag) - 2
|
||
if max_prompt > 4 and len(prompt_line) > max_prompt:
|
||
prompt_line = prompt_line[:max_prompt] + "…"
|
||
return f"{size_tag}{prompt_line}"
|
||
|
||
|
||
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
|
||
|
||
_set_requesting(self, True)
|
||
is_remote_url = _svc.is_remote_image_url(prompt)
|
||
self.ai_image_status_var.set("下载中…" if is_remote_url else "请求中…")
|
||
|
||
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))
|
||
|
||
if is_remote_url:
|
||
_svc.import_image_from_url_async(
|
||
prompt,
|
||
on_success=_success,
|
||
on_error=_error,
|
||
)
|
||
return
|
||
|
||
if not _svc.has_api():
|
||
_set_requesting(self, False)
|
||
self.ai_image_status_var.set("就绪")
|
||
messagebox.showerror(
|
||
"API 未配置",
|
||
"AI 图片 API 尚未接入。\n"
|
||
"可直接输入图片 URL 导入，或在启动时通过 "
|
||
"app.services.ai_image.set_api_caller(...) 注入真实实现。",
|
||
)
|
||
return
|
||
|
||
_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, exc):
|
||
_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)
|
||
|
||
|
||
def _rename_current(self):
|
||
"""弹窗让用户修改当前记录的备注名称（即列表显示中 size_tag 之后的部分）。"""
|
||
rec = getattr(self, "ai_image_current", None)
|
||
if rec is None:
|
||
messagebox.showinfo("提示", "请先选择一张图片")
|
||
return
|
||
|
||
current_name = rec.prompt or ""
|
||
new_name = simpledialog.askstring(
|
||
"重命名",
|
||
"修改备注名称（显示在分辨率标签后面）：",
|
||
initialvalue=current_name,
|
||
parent=self.root,
|
||
)
|
||
if new_name is None: # 用户点了取消
|
||
return
|
||
new_name = new_name.strip()
|
||
if not new_name:
|
||
messagebox.showwarning("提示", "备注名称不能为空")
|
||
return
|
||
if new_name == current_name:
|
||
return
|
||
|
||
# 写回 JSON 元数据
|
||
try:
|
||
import json
|
||
meta_path = os.path.splitext(rec.image_path)[0] + ".json"
|
||
meta = {}
|
||
if os.path.isfile(meta_path):
|
||
with open(meta_path, "r", encoding="utf-8") as f:
|
||
meta = json.load(f)
|
||
meta["prompt"] = new_name
|
||
with open(meta_path, "w", encoding="utf-8") as f:
|
||
json.dump(meta, f, ensure_ascii=False, indent=2)
|
||
except Exception as exc:
|
||
messagebox.showerror("保存失败", f"无法更新元数据:\n{exc}")
|
||
return
|
||
|
||
# 同步内存中的记录并刷新列表
|
||
rec.prompt = new_name
|
||
reload_ai_image_list(self)
|
||
# 重新定位到刚才被重命名的图片
|
||
for i, r in enumerate(self.ai_image_records):
|
||
if r.id == rec.id:
|
||
self.ai_image_listbox.selection_clear(0, tk.END)
|
||
self.ai_image_listbox.selection_set(i)
|
||
self.ai_image_listbox.activate(i)
|
||
self.ai_image_listbox.see(i)
|
||
_select_record(self, r)
|
||
break
|
||
|
||
|
||
# ---------------- 发送到 UCD ----------------
|
||
|
||
|
||
def _show_list_context_menu(self, event):
|
||
"""在图片列表上显示右键菜单，并根据状态启用/禁用项。"""
|
||
try:
|
||
idx = self.ai_image_listbox.nearest(event.y)
|
||
except Exception:
|
||
idx = -1
|
||
if 0 <= idx < len(self.ai_image_records):
|
||
self.ai_image_listbox.selection_clear(0, tk.END)
|
||
self.ai_image_listbox.selection_set(idx)
|
||
self.ai_image_listbox.activate(idx)
|
||
_select_record(self, self.ai_image_records[idx])
|
||
|
||
has_selection = self.ai_image_current is not None
|
||
ucd = getattr(self, "ucd", None)
|
||
can_send = (
|
||
has_selection
|
||
and ucd is not None
|
||
and getattr(ucd, "status", False)
|
||
)
|
||
try:
|
||
self.ai_image_menu.entryconfigure(
|
||
0, state=("normal" if can_send else "disabled")
|
||
)
|
||
self.ai_image_menu.entryconfigure(
|
||
2, state=("normal" if has_selection else "disabled")
|
||
)
|
||
self.ai_image_menu.entryconfigure(
|
||
3, state=("normal" if has_selection else "disabled")
|
||
)
|
||
self.ai_image_menu.entryconfigure(
|
||
4, state=("normal" if has_selection else "disabled")
|
||
)
|
||
self.ai_image_menu.tk_popup(event.x_root, event.y_root)
|
||
finally:
|
||
self.ai_image_menu.grab_release()
|
||
|
||
|
||
def _send_to_ucd(self):
|
||
"""把当前选中的 AI 图片通过 UCD 发送到显示设备。"""
|
||
rec = getattr(self, "ai_image_current", None)
|
||
if rec is None:
|
||
messagebox.showinfo("提示", "请先选择一张图片")
|
||
return
|
||
if not os.path.isfile(rec.image_path):
|
||
messagebox.showerror("错误", f"图片文件不存在:\n{rec.image_path}")
|
||
return
|
||
ucd = getattr(self, "ucd", None)
|
||
if ucd is None or not getattr(ucd, "status", False):
|
||
messagebox.showwarning("警告", "请先连接 UCD323 设备")
|
||
return
|
||
|
||
image_path = rec.image_path
|
||
send_path = image_path
|
||
log = getattr(self, "log_gui", None)
|
||
|
||
# 发送前检查分辨率：建议与当前 UCD 输出分辨率一致。
|
||
try:
|
||
target_w, target_h = get_current_resolution(ucd)
|
||
except Exception:
|
||
target_w, target_h = 3840, 2160
|
||
try:
|
||
with Image.open(image_path) as _img:
|
||
src_w, src_h = _img.size
|
||
except Exception as exc:
|
||
messagebox.showerror("错误", f"无法读取图片尺寸:\n{exc}")
|
||
return
|
||
|
||
if log is not None:
|
||
log.log(
|
||
f"UCD分辨率: {target_w}x{target_h} | 图片分辨率: {src_w}x{src_h}",
|
||
level="info",
|
||
)
|
||
|
||
if (src_w, src_h) != (target_w, target_h):
|
||
if not messagebox.askyesno(
|
||
"分辨率不匹配",
|
||
(
|
||
f"当前 UCD 分辨率: {target_w}x{target_h}\n"
|
||
f"图片分辨率: {src_w}x{src_h}\n\n"
|
||
"是否自动缩放后再发送？"
|
||
),
|
||
):
|
||
if log is not None:
|
||
log.log("用户取消发送：图片分辨率与 UCD 不一致", level="warning")
|
||
return
|
||
|
||
try:
|
||
send_path = _build_ucd_resized_image(image_path, target_w, target_h)
|
||
if log is not None:
|
||
log.log(
|
||
f"已生成匹配分辨率副本: {os.path.basename(send_path)}",
|
||
level="info",
|
||
)
|
||
except Exception as exc:
|
||
messagebox.showerror("错误", f"自动缩放失败:\n{exc}")
|
||
return
|
||
|
||
if log is not None:
|
||
log.log(
|
||
f"发送 AI 图片到 UCD: {os.path.basename(send_path)}",
|
||
level="info",
|
||
)
|
||
if hasattr(self, "ai_image_status_var"):
|
||
self.ai_image_status_var.set("发送中…")
|
||
|
||
def _worker():
|
||
err = None
|
||
try:
|
||
ok = send_image_pattern(ucd, send_path)
|
||
except Exception as exc:
|
||
ok = False
|
||
err = str(exc)
|
||
|
||
def _done():
|
||
if ok:
|
||
if log is not None:
|
||
log.log(
|
||
f"图片已发送到 UCD: {os.path.basename(send_path)}",
|
||
level="success",
|
||
)
|
||
if hasattr(self, "ai_image_status_var"):
|
||
self.ai_image_status_var.set("已发送到 UCD")
|
||
else:
|
||
msg = f"UCD 发送失败: {err}" if err else "UCD 发送失败"
|
||
if log is not None:
|
||
log.log(msg, level="error")
|
||
if hasattr(self, "ai_image_status_var"):
|
||
self.ai_image_status_var.set("UCD 发送失败")
|
||
|
||
try:
|
||
self.root.after(0, _done)
|
||
except Exception:
|
||
pass
|
||
|
||
threading.Thread(target=_worker, daemon=True).start()
|
||
|
||
|
||
def _build_ucd_resized_image(image_path: str, target_w: int, target_h: int) -> str:
|
||
"""生成与 UCD 分辨率匹配的临时副本（缓存目录内）。"""
|
||
base_dir = os.path.dirname(image_path)
|
||
base_name = os.path.splitext(os.path.basename(image_path))[0]
|
||
out_name = f"{base_name}_{target_w}x{target_h}_ucd.png"
|
||
out_path = os.path.join(base_dir, out_name)
|
||
with Image.open(image_path) as img:
|
||
resized = img.convert("RGB").resize((target_w, target_h), Image.LANCZOS)
|
||
resized.save(out_path, format="PNG")
|
||
return out_path
|