2271 lines
97 KiB
Python
2271 lines
97 KiB
Python
# coding=utf-8
|
||
"""菜品识别公共 UI：模板管理、识别、待学习、API 上报、同步。
|
||
|
||
海康 / USB 入口只需继承本类，实现相机连接、拍照、关闭。
|
||
|
||
界面结构（结算机）::
|
||
|
||
┌ 顶栏：标题 + 状态徽章 + 设置 ─────────────────────┐
|
||
├ 左：已点菜品清单 ─────────┬─ 右：绿提示 + 大预览 ─┤
|
||
│ 缩略图 / 菜名 / 数量 │ │ 实时视频流（约 1fps） │
|
||
├──────────────────────────┴───────────────────────┤
|
||
│ 底栏：暂停/继续扫描｜选图识别｜清空｜设置 │
|
||
└──────────────────────────────────────────────────┘
|
||
"""
|
||
import json
|
||
import os
|
||
import sys
|
||
import threading
|
||
import time
|
||
from pathlib import Path
|
||
|
||
import cv2
|
||
import numpy as np
|
||
import tkinter as tk
|
||
from tkinter import ttk, filedialog, messagebox
|
||
from PIL import Image, ImageTk
|
||
|
||
from core.detect_plate_process import Config, YOLOv11PlateExtractor
|
||
from core.common_classifier import PartTemplateManager, TEMPLATE_SAVE_DIR, THUMBNAIL_DIR
|
||
from ui.pending_learning_ui import PendingLearningUIMixin
|
||
from ui.theme import COLORS, FONTS, RADIUS, SIZES, apply_theme
|
||
from ui.widgets import (
|
||
StatusBadge, NoticeBar, IconButton, RoundedPreview, RoundCard, PillButton,
|
||
SectionTitle, SearchBox, Chip, EmptyState, Divider, Toast, StatTile,
|
||
make_rounded_thumb, make_photo_rounded,
|
||
)
|
||
from service.result_api import ResultApiClient, default_media_upload_url
|
||
from service.device_status import DeviceStatusReporter
|
||
from dao.database import is_server_available, local_session_scope
|
||
from dao.recognition_repository import RecognitionRepository
|
||
from dao.meal_schedule import get_meal_schedule_store
|
||
from dao.schedule_check import flag_dishes_not_in_menu
|
||
from dao.settings import DISH_BASE_NAME, DISH_MGMT_MEAL_ON_START
|
||
from models.App_settings import AppSetting
|
||
from dao.template_sync import SETTING_LAST_SYNC
|
||
|
||
os.makedirs(TEMPLATE_SAVE_DIR, exist_ok=True)
|
||
os.makedirs(THUMBNAIL_DIR, exist_ok=True)
|
||
|
||
_PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
_MOCK_DIR = os.path.join(_PROJECT_ROOT, "api_mock_results")
|
||
|
||
|
||
class BaseDishApp(PendingLearningUIMixin):
|
||
"""公共界面与业务逻辑；子类实现相机相关方法。"""
|
||
|
||
window_title = "菜品识别系统"
|
||
app_title = "菜品识别系统"
|
||
edition_label = "标准版"
|
||
camera_display_name = "相机"
|
||
logo_glyph = "菜"
|
||
|
||
def __init__(self, root):
|
||
self.root = root
|
||
self.root.title(self.window_title)
|
||
self.root.state('zoomed')
|
||
self.root.encoding = 'utf-8'
|
||
|
||
apply_theme(self.root)
|
||
|
||
self.manager = PartTemplateManager(0.7)
|
||
self.current_img = None
|
||
self.detect_img = None
|
||
self.camera_connected = False
|
||
base_info = self.manager.get_base_info()
|
||
self.result_api = ResultApiClient(
|
||
base_id=base_info.get('base_code'),
|
||
base_name=base_info.get('base_name') or DISH_BASE_NAME,
|
||
)
|
||
self.recognition_repo = RecognitionRepository()
|
||
self.meal_schedule = get_meal_schedule_store()
|
||
self.device_status = DeviceStatusReporter()
|
||
|
||
# 尽早拉起客户端后台服务（与后续 YOLO/UI 加载并行）
|
||
threading.Thread(target=self._bootstrap_mgmt_services, daemon=True).start()
|
||
|
||
self.class_name = tk.StringVar(value="")
|
||
self.thresh_val = tk.DoubleVar(value=self.manager.threshold)
|
||
self._selected_dish = None
|
||
self._thumb_refs = []
|
||
self._dock_thumb_refs = []
|
||
self._cart = []
|
||
self._cart_thumbs = []
|
||
self._template_filter = ""
|
||
self._scroll_canvases = []
|
||
self._wheel_bound = False
|
||
self._last_dishes = []
|
||
self._live_enabled = False
|
||
self._live_job = None
|
||
self._recognize_busy = False
|
||
self._live_lock = threading.Lock()
|
||
self._latest_frame = None # BGR ndarray
|
||
self._live_interval_ms = 1000 # 约 1 秒一帧取图识别
|
||
# 实时：结果稳定满 N 秒后，才把未知菜写入待学习（同场景只写一次）
|
||
self._live_stable_seconds = 5.0
|
||
self._live_stable_sig = None
|
||
self._live_stable_since = None
|
||
self._live_stable_committed_sig = None
|
||
self._live_stable_snapshot = None # 稳定期间保留的 dishes，供写入
|
||
self.cfg = Config()
|
||
self.app = YOLOv11PlateExtractor(self.cfg)
|
||
self.final_result = ""
|
||
|
||
self.create_ui()
|
||
self.auto_connect_camera()
|
||
# 后台同步后刷新基地/同步状态
|
||
self.manager.repo.set_reload_callback(self._on_templates_reloaded)
|
||
|
||
def _bootstrap_mgmt_services(self):
|
||
print('[INFO] 启动客户端服务：媒体HTTP / 排餐 / 设备状态…')
|
||
try:
|
||
from service.media_server import ensure_media_server
|
||
ensure_media_server()
|
||
except Exception as exc:
|
||
print(f'[WARN] media server bootstrap failed: {exc}')
|
||
try:
|
||
if DISH_MGMT_MEAL_ON_START:
|
||
self.meal_schedule.refresh(base_name=DISH_BASE_NAME, force=True)
|
||
self.meal_schedule.start_morning_refresh(base_name=DISH_BASE_NAME)
|
||
except Exception as exc:
|
||
print(f'[WARN] meal schedule bootstrap failed: {exc}')
|
||
try:
|
||
self.device_status.start()
|
||
except Exception as exc:
|
||
print(f'[WARN] device status reporter start failed: {exc}')
|
||
print('[OK] 客户端服务已就绪')
|
||
|
||
def shutdown_mgmt_services(self):
|
||
try:
|
||
self.device_status.stop()
|
||
except Exception:
|
||
pass
|
||
try:
|
||
self.meal_schedule.stop_morning_refresh()
|
||
except Exception:
|
||
pass
|
||
try:
|
||
from service.media_server import stop_media_server
|
||
stop_media_server()
|
||
except Exception:
|
||
pass
|
||
|
||
def grab_preview_frame(self):
|
||
"""子类实现：抓取一帧 BGR 图像，失败返回 None。"""
|
||
return None
|
||
|
||
def take_live_frame(self):
|
||
"""抓取实时帧；返回 (0, path, frame) 或 (-1, None, None)。
|
||
|
||
优先内存帧供 YOLO；磁盘路径供上报/待学习（需要时再写）。
|
||
"""
|
||
frame = self.grab_preview_frame()
|
||
if frame is None:
|
||
return -1, None, None
|
||
os.makedirs("images", exist_ok=True)
|
||
stamp = int(time.time() * 1000)
|
||
save_path = os.path.abspath(os.path.join("images", f"_live_frame_{stamp}.jpg"))
|
||
try:
|
||
if cv2.imwrite(save_path, frame):
|
||
self._cleanup_old_live_frames(keep=30)
|
||
return 0, save_path, frame
|
||
except Exception as e:
|
||
print(f"保存实时帧失败: {e}")
|
||
return -1, None, None
|
||
|
||
def _cleanup_old_live_frames(self, keep=30):
|
||
"""清理过期实时帧，防止 images/ 无限增长。"""
|
||
try:
|
||
folder = os.path.abspath("images")
|
||
files = []
|
||
for name in os.listdir(folder):
|
||
if not (name.startswith("_live_frame_") and name.endswith(".jpg")):
|
||
continue
|
||
path = os.path.join(folder, name)
|
||
if os.path.isfile(path):
|
||
files.append(path)
|
||
if len(files) <= keep:
|
||
return
|
||
files.sort(key=lambda p: os.path.getmtime(p))
|
||
for path in files[:-keep]:
|
||
try:
|
||
os.remove(path)
|
||
except Exception:
|
||
pass
|
||
except Exception:
|
||
pass
|
||
|
||
def _widget_alive(self, widget):
|
||
try:
|
||
return widget is not None and bool(widget.winfo_exists())
|
||
except Exception:
|
||
return False
|
||
|
||
def _on_templates_reloaded(self):
|
||
try:
|
||
self.manager.load_templates()
|
||
except Exception:
|
||
pass
|
||
try:
|
||
self.root.after(0, self.refresh_template_list)
|
||
self.root.after(0, self.refresh_base_status)
|
||
except Exception:
|
||
pass
|
||
|
||
# ---------- 子类必须实现 ----------
|
||
def auto_connect_camera(self):
|
||
raise NotImplementedError
|
||
|
||
def take_photo(self):
|
||
"""返回 (0, path) 成功；(-1, None/错误信息) 失败。"""
|
||
raise NotImplementedError
|
||
|
||
def close_camera(self):
|
||
raise NotImplementedError
|
||
|
||
# ---------------------------------------------------------------
|
||
# 通用小工具
|
||
# ---------------------------------------------------------------
|
||
def _toast(self, text, tone="info"):
|
||
try:
|
||
Toast.show(self.root, text, tone=tone)
|
||
except Exception:
|
||
pass
|
||
|
||
def _make_scroll_area(self, parent, bg):
|
||
"""返回 (wrapper, inner_frame, canvas)，并注册滚轮支持。"""
|
||
wrap = tk.Frame(parent, bg=bg)
|
||
canvas = tk.Canvas(wrap, bg=bg, highlightthickness=0, bd=0)
|
||
bar = ttk.Scrollbar(wrap, orient="vertical", command=canvas.yview)
|
||
inner = tk.Frame(canvas, bg=bg)
|
||
inner.bind("<Configure>", lambda e: canvas.configure(scrollregion=canvas.bbox("all")))
|
||
win = canvas.create_window((0, 0), window=inner, anchor="nw")
|
||
canvas.configure(yscrollcommand=bar.set)
|
||
canvas.bind("<Configure>", lambda e: canvas.itemconfigure(win, width=max(e.width, 1)))
|
||
canvas.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
|
||
bar.pack(side=tk.RIGHT, fill=tk.Y)
|
||
self._register_scroll(canvas)
|
||
return wrap, inner, canvas
|
||
|
||
def _register_scroll(self, canvas):
|
||
"""统一接管滚轮：按指针位置决定滚动哪个滚动区。"""
|
||
if canvas not in self._scroll_canvases:
|
||
self._scroll_canvases.append(canvas)
|
||
if not self._wheel_bound:
|
||
self._wheel_bound = True
|
||
self.root.bind_all("<MouseWheel>", self._on_wheel_all)
|
||
|
||
def _on_wheel_all(self, event):
|
||
alive = []
|
||
for canvas in list(getattr(self, "_scroll_canvases", [])):
|
||
try:
|
||
if not canvas.winfo_exists():
|
||
continue
|
||
alive.append(canvas)
|
||
x, y = canvas.winfo_rootx(), canvas.winfo_rooty()
|
||
if x <= event.x_root <= x + canvas.winfo_width() and \
|
||
y <= event.y_root <= y + canvas.winfo_height():
|
||
canvas.yview_scroll(-1 if event.delta > 0 else 1, "units")
|
||
return
|
||
except Exception:
|
||
continue
|
||
self._scroll_canvases = alive
|
||
|
||
# ---------- API / 识别落库 ----------
|
||
def report_recognition(self, source_image, dishes, labeled_image=None, schedule_check=True):
|
||
# 已识别但不在今日排餐 → 待学习库（人工确认是否识错）
|
||
# 实时扫描关闭 schedule_check，避免每秒把「不在菜单」刷进待学习
|
||
if schedule_check:
|
||
try:
|
||
flag_dishes_not_in_menu(self.manager, dishes, self.meal_schedule)
|
||
except Exception as e:
|
||
print(f"[MEAL] schedule check failed: {e}")
|
||
|
||
payload = None
|
||
try:
|
||
payload = self.result_api.send(
|
||
source_image=source_image,
|
||
dishes=dishes,
|
||
labeled_image=labeled_image,
|
||
async_mode=True,
|
||
)
|
||
except Exception as e:
|
||
print(f"[API] report failed: {e}")
|
||
|
||
try:
|
||
base_info = self.manager.get_base_info()
|
||
request_id = None
|
||
capture_time = None
|
||
if isinstance(payload, dict):
|
||
request_id = payload.get('request_id')
|
||
capture_time = payload.get('capture_time')
|
||
self.recognition_repo.save(
|
||
source_image=source_image,
|
||
dishes=dishes,
|
||
labeled_image=labeled_image,
|
||
request_id=request_id,
|
||
base_code=base_info.get('base_code'),
|
||
source_base_id=base_info.get('id'),
|
||
capture_time=capture_time,
|
||
)
|
||
except Exception as e:
|
||
print(f"[DB] recognition save failed: {e}")
|
||
|
||
def _format_base_sync_status(self):
|
||
info = self.manager.get_base_info()
|
||
code = info.get('base_code') or '—'
|
||
online = is_server_available()
|
||
last_sync = None
|
||
pending = failed = 0
|
||
try:
|
||
st = self.manager.get_sync_status()
|
||
last_sync = st.get('last_sync')
|
||
pending = int(st.get('pending') or 0)
|
||
failed = int(st.get('failed') or 0)
|
||
online = bool(st.get('online')) if 'online' in st else online
|
||
except Exception:
|
||
try:
|
||
with local_session_scope() as session:
|
||
row = session.get(AppSetting, SETTING_LAST_SYNC)
|
||
last_sync = row.value if row and row.value else None
|
||
except Exception:
|
||
last_sync = None
|
||
sync_short = last_sync[-8:] if last_sync else '—'
|
||
q_part = ''
|
||
if pending or failed:
|
||
q_part = f' · 队{pending}'
|
||
if failed:
|
||
q_part += f'/败{failed}'
|
||
if online:
|
||
text = f"{code} · 在线 {sync_short}{q_part}"
|
||
tone = "warn" if failed else "mute"
|
||
else:
|
||
text = f"{code} · 离线 {sync_short}{q_part}"
|
||
tone = "warn"
|
||
if hasattr(self, 'base_badge'):
|
||
self.base_badge.set(text, tone)
|
||
return text
|
||
|
||
def refresh_base_status(self):
|
||
self._format_base_sync_status()
|
||
|
||
def sync_templates_now(self):
|
||
"""手动立即同步（后台线程）。"""
|
||
if getattr(self, '_syncing', False):
|
||
self._set_result_text("同步进行中…", "busy")
|
||
return
|
||
self._syncing = True
|
||
self._set_result_text("正在同步模板库…", "busy")
|
||
|
||
def work():
|
||
try:
|
||
result = self.manager.sync_now(full=True)
|
||
except Exception as e:
|
||
result = {'ok': False, 'message': str(e)}
|
||
|
||
def done():
|
||
self._syncing = False
|
||
try:
|
||
self.manager.load_templates()
|
||
except Exception:
|
||
pass
|
||
self.refresh_template_list()
|
||
self.refresh_base_status()
|
||
msg = result.get('message') or ('同步完成' if result.get('ok') else '同步失败')
|
||
self._set_result_text(msg, "ok" if result.get('ok') else "err")
|
||
if hasattr(self, '_sync_dialog') and self._sync_dialog and self._sync_dialog.winfo_exists():
|
||
self._refresh_sync_dialog()
|
||
|
||
self.root.after(0, done)
|
||
|
||
threading.Thread(target=work, daemon=True).start()
|
||
|
||
def retry_upload_queue(self):
|
||
if getattr(self, '_syncing', False):
|
||
return
|
||
self._syncing = True
|
||
self._set_result_text("正在重试失败上传…", "busy")
|
||
|
||
def work():
|
||
try:
|
||
result = self.manager.retry_failed_uploads()
|
||
except Exception as e:
|
||
result = {'ok': False, 'message': str(e)}
|
||
|
||
def done():
|
||
self._syncing = False
|
||
try:
|
||
self.manager.load_templates()
|
||
except Exception:
|
||
pass
|
||
self.refresh_template_list()
|
||
self.refresh_base_status()
|
||
msg = result.get('message') or '重试完成'
|
||
self._set_result_text(msg, "ok" if result.get('ok') else "warn")
|
||
if hasattr(self, '_sync_dialog') and self._sync_dialog and self._sync_dialog.winfo_exists():
|
||
self._refresh_sync_dialog()
|
||
|
||
self.root.after(0, done)
|
||
|
||
threading.Thread(target=work, daemon=True).start()
|
||
|
||
# ---------------------------------------------------------------
|
||
# 弹窗：同步 / 记录 / 设置 / 统计
|
||
# ---------------------------------------------------------------
|
||
def _dialog_shell(self, title, width, height, grab=True):
|
||
"""统一的新窗口外壳：卡片式浅色底，居中显示。"""
|
||
dialog = tk.Toplevel(self.root)
|
||
dialog.title(title)
|
||
dialog.configure(bg=COLORS["bg"])
|
||
self.center_window(dialog, width, height)
|
||
dialog.transient(self.root)
|
||
if grab:
|
||
dialog.grab_set()
|
||
return dialog
|
||
|
||
def show_sync_dialog(self):
|
||
"""同步状态 / 上传队列面板。"""
|
||
dialog = self._dialog_shell("模板同步", 820, 560)
|
||
self._sync_dialog = dialog
|
||
|
||
top = tk.Frame(dialog, bg=COLORS["bg"])
|
||
top.pack(fill=tk.X, padx=14, pady=(12, 6))
|
||
self._sync_status_label = tk.Label(
|
||
top, text="", bg=COLORS["bg"], fg=COLORS["text_body"], font=FONTS["ui"], anchor="w",
|
||
)
|
||
self._sync_status_label.pack(side=tk.LEFT, fill=tk.X, expand=True)
|
||
|
||
btns = tk.Frame(top, bg=COLORS["bg"])
|
||
btns.pack(side=tk.RIGHT)
|
||
PillButton(btns, text="立即同步", variant="primary", size="sm",
|
||
command=self.sync_templates_now).pack(side=tk.LEFT, padx=4)
|
||
PillButton(btns, text="重试失败", variant="secondary", size="sm",
|
||
command=self.retry_upload_queue).pack(side=tk.LEFT, padx=4)
|
||
PillButton(btns, text="刷新", variant="ghost", size="sm",
|
||
command=self._refresh_sync_dialog).pack(side=tk.LEFT, padx=4)
|
||
|
||
tk.Label(
|
||
dialog,
|
||
text="离线学菜 / 删菜会进入上传队列；联网后可立即同步，或等待后台定时同步。",
|
||
bg=COLORS["bg"], fg=COLORS["muted"], font=FONTS["small"],
|
||
).pack(anchor=tk.W, padx=16)
|
||
|
||
table_card = RoundCard(dialog, radius=RADIUS["lg"], padding=10, expand_body=True)
|
||
table_card.pack(fill=tk.BOTH, expand=True, padx=12, pady=10)
|
||
|
||
cols = ("id", "status", "action", "retry", "time", "error")
|
||
tree = ttk.Treeview(table_card.body, columns=cols, show="headings", height=14)
|
||
tree.heading("id", text="ID")
|
||
tree.heading("status", text="状态")
|
||
tree.heading("action", text="动作")
|
||
tree.heading("retry", text="重试")
|
||
tree.heading("time", text="更新时间")
|
||
tree.heading("error", text="错误")
|
||
tree.column("id", width=50, anchor=tk.CENTER)
|
||
tree.column("status", width=70, anchor=tk.CENTER)
|
||
tree.column("action", width=110)
|
||
tree.column("retry", width=50, anchor=tk.CENTER)
|
||
tree.column("time", width=140)
|
||
tree.column("error", width=280)
|
||
tree.pack(fill=tk.BOTH, expand=True)
|
||
self._sync_tree = tree
|
||
|
||
PillButton(dialog, text="关闭", variant="ghost",
|
||
command=dialog.destroy).pack(pady=(0, 12))
|
||
self._refresh_sync_dialog()
|
||
|
||
def _refresh_sync_dialog(self):
|
||
if not hasattr(self, '_sync_status_label'):
|
||
return
|
||
try:
|
||
st = self.manager.get_sync_status()
|
||
except Exception as e:
|
||
self._sync_status_label.config(text=f"读取失败：{e}")
|
||
return
|
||
online = '在线' if st.get('online') else '离线'
|
||
err = st.get('last_error') or ''
|
||
err_txt = f"｜最近错误：{err}" if err else ''
|
||
self._sync_status_label.config(
|
||
text=(
|
||
f"总库{online}｜上次同步 {st.get('last_sync') or '—'}｜"
|
||
f"待上传 {st.get('pending', 0)}｜失败 {st.get('failed', 0)}{err_txt}"
|
||
)
|
||
)
|
||
tree = getattr(self, '_sync_tree', None)
|
||
if tree is None:
|
||
return
|
||
for item in tree.get_children():
|
||
tree.delete(item)
|
||
action_cn = {
|
||
'insert': '新增模板',
|
||
'delete_template': '删模板',
|
||
'delete_category': '删类别',
|
||
'clear_all': '清空',
|
||
'set_threshold': '改阈值',
|
||
}
|
||
try:
|
||
rows = self.manager.list_upload_queue(statuses=('pending', 'failed'), limit=120)
|
||
except Exception:
|
||
rows = []
|
||
for row in rows:
|
||
tree.insert(
|
||
'',
|
||
tk.END,
|
||
values=(
|
||
row.get('id'),
|
||
row.get('status'),
|
||
action_cn.get(row.get('action'), row.get('action')),
|
||
row.get('retry_count') or 0,
|
||
row.get('update_time') or '',
|
||
(row.get('error_msg') or '')[:80],
|
||
),
|
||
)
|
||
self.refresh_base_status()
|
||
|
||
def show_recognition_history(self):
|
||
"""查看最近识别记录（本地库）。"""
|
||
try:
|
||
records = self.recognition_repo.list_recent(limit=80)
|
||
except Exception as e:
|
||
messagebox.showerror("错误", f"读取识别记录失败：{e}")
|
||
return
|
||
|
||
dialog = self._dialog_shell("识别记录", 880, 580)
|
||
|
||
header = tk.Frame(dialog, bg=COLORS["bg"])
|
||
header.pack(fill=tk.X, padx=14, pady=(12, 6))
|
||
tk.Label(
|
||
header,
|
||
text=f"最近 {len(records)} 条（本地 SQLite；MySQL 可用时会同步写入总库）",
|
||
bg=COLORS["bg"], fg=COLORS["text_body"], font=FONTS["ui"],
|
||
).pack(side=tk.LEFT)
|
||
PillButton(header, text="识别统计", variant="secondary", size="sm",
|
||
command=self.show_statistics_dialog).pack(side=tk.RIGHT)
|
||
|
||
if not records:
|
||
card = RoundCard(dialog, radius=RADIUS["lg"], padding=10, expand_body=True)
|
||
card.pack(fill=tk.BOTH, expand=True, padx=12, pady=10)
|
||
EmptyState(card.body, text="暂无识别记录", hint="完成一次选图或拍照识别后，记录会出现在这里",
|
||
glyph="◍", bg=COLORS["surface"]).pack(fill=tk.BOTH, expand=True)
|
||
PillButton(dialog, text="关闭", variant="ghost",
|
||
command=dialog.destroy).pack(pady=(0, 12))
|
||
return
|
||
|
||
card = RoundCard(dialog, radius=RADIUS["lg"], padding=10, expand_body=True)
|
||
card.pack(fill=tk.BOTH, expand=True, padx=12, pady=(0, 4))
|
||
|
||
cols = ("id", "time", "base", "count", "unknown", "summary")
|
||
tree = ttk.Treeview(card.body, columns=cols, show="headings", height=18)
|
||
tree.heading("id", text="ID")
|
||
tree.heading("time", text="识别时间")
|
||
tree.heading("base", text="基地")
|
||
tree.heading("count", text="菜数")
|
||
tree.heading("unknown", text="未知")
|
||
tree.heading("summary", text="摘要")
|
||
tree.column("id", width=50, anchor=tk.CENTER)
|
||
tree.column("time", width=150, anchor=tk.CENTER)
|
||
tree.column("base", width=110, anchor=tk.CENTER)
|
||
tree.column("count", width=50, anchor=tk.CENTER)
|
||
tree.column("unknown", width=50, anchor=tk.CENTER)
|
||
tree.column("summary", width=400, anchor=tk.W)
|
||
tree.pack(fill=tk.BOTH, expand=True)
|
||
|
||
tk.Label(dialog, text="双击某一行查看完整结果 JSON",
|
||
bg=COLORS["bg"], fg=COLORS["muted"], font=FONTS["small"],
|
||
).pack(anchor=tk.W, padx=16, pady=(2, 0))
|
||
|
||
for r in records:
|
||
tree.insert(
|
||
"",
|
||
tk.END,
|
||
iid=str(r["id"]),
|
||
values=(
|
||
r.get("id"),
|
||
r.get("capture_time") or "",
|
||
r.get("base_code") or "",
|
||
r.get("dish_count") or 0,
|
||
r.get("unknown_count") or 0,
|
||
r.get("summary") or "",
|
||
),
|
||
)
|
||
|
||
def _on_double(_event=None):
|
||
sel = tree.selection()
|
||
if not sel:
|
||
return
|
||
rid = int(sel[0])
|
||
row = self.recognition_repo.get(rid)
|
||
if not row:
|
||
return
|
||
win = self._dialog_shell(f"识别详情 #{rid}", 660, 520, grab=False)
|
||
text = tk.Text(
|
||
win, wrap=tk.WORD, bg=COLORS["surface"], fg=COLORS["text_body"],
|
||
font=FONTS["mono"], relief=tk.FLAT, highlightthickness=1,
|
||
highlightbackground=COLORS["border"], padx=10, pady=10,
|
||
)
|
||
text.pack(fill=tk.BOTH, expand=True, padx=12, pady=12)
|
||
try:
|
||
pretty = json.dumps(json.loads(row.get("result_json") or "[]"),
|
||
ensure_ascii=False, indent=2)
|
||
except Exception:
|
||
pretty = row.get("result_json") or ""
|
||
body = (
|
||
f"request_id: {row.get('request_id')}\n"
|
||
f"时间: {row.get('capture_time')}\n"
|
||
f"基地: {row.get('base_code')}\n"
|
||
f"原图: {row.get('source_image')}\n"
|
||
f"标注图: {row.get('labeled_image')}\n"
|
||
f"摘要: {row.get('summary')}\n\n"
|
||
f"{pretty}"
|
||
)
|
||
text.insert("1.0", body)
|
||
text.config(state=tk.DISABLED)
|
||
PillButton(win, text="关闭", variant="ghost", command=win.destroy).pack(pady=(0, 12))
|
||
|
||
tree.bind("<Double-1>", _on_double)
|
||
PillButton(dialog, text="关闭", variant="ghost",
|
||
command=dialog.destroy).pack(pady=(0, 12))
|
||
|
||
# ---------- 设置 ----------
|
||
def show_settings_dialog(self):
|
||
"""设置工作台：学菜 / 模板库 / 阈值 / 上报 / 同步入口。"""
|
||
existing = getattr(self, "_settings_dialog", None)
|
||
if self._widget_alive(existing):
|
||
try:
|
||
existing.lift()
|
||
existing.focus_force()
|
||
except Exception:
|
||
pass
|
||
return
|
||
|
||
dialog = self._dialog_shell("设置", 980, 700)
|
||
self._settings_dialog = dialog
|
||
|
||
def _close_settings():
|
||
for attr in (
|
||
"scrollable_frame", "_template_canvas", "template_stats",
|
||
"search_box", "thresh_scale",
|
||
):
|
||
setattr(self, attr, None)
|
||
self._settings_dialog = None
|
||
try:
|
||
dialog.destroy()
|
||
except Exception:
|
||
pass
|
||
|
||
dialog.protocol("WM_DELETE_WINDOW", _close_settings)
|
||
nb = ttk.Notebook(dialog)
|
||
nb.pack(fill=tk.BOTH, expand=True, padx=10, pady=10)
|
||
|
||
# ---- 学菜 ----
|
||
tab_learn = tk.Frame(nb, bg=COLORS["bg"])
|
||
nb.add(tab_learn, text="模板学习")
|
||
self._build_learn_panel(tab_learn, COLORS)
|
||
|
||
# ---- 模板库 ----
|
||
tab_tpl = tk.Frame(nb, bg=COLORS["sidebar"])
|
||
nb.add(tab_tpl, text="模板库")
|
||
self._build_sidebar(tab_tpl, COLORS)
|
||
self.refresh_template_list()
|
||
|
||
# ---- 识别与上报 ----
|
||
tab_cfg = tk.Frame(nb, bg=COLORS["bg"])
|
||
nb.add(tab_cfg, text="识别与上报")
|
||
|
||
tk.Label(tab_cfg, text="识别阈值", bg=COLORS["bg"], fg=COLORS["text"],
|
||
font=FONTS["h2"]).pack(anchor=tk.W, padx=18, pady=(16, 4))
|
||
tk.Label(tab_cfg, text="相似度低于该值判定为未知。实时扫描：未知不入清单；画面稳定约 5 秒后才写入待学习。选图/拍照仍即时入待学习。",
|
||
bg=COLORS["bg"], fg=COLORS["muted"], font=FONTS["small"],
|
||
).pack(anchor=tk.W, padx=18)
|
||
|
||
thresh_row = tk.Frame(tab_cfg, bg=COLORS["bg"])
|
||
thresh_row.pack(fill=tk.X, padx=18, pady=(8, 0))
|
||
local_thresh = tk.DoubleVar(value=self.manager.threshold)
|
||
ttk.Spinbox(thresh_row, from_=0.0, to=1.0, increment=0.01, width=6,
|
||
textvariable=local_thresh, style="Large.TSpinbox").pack(side=tk.LEFT)
|
||
scale = ttk.Scale(thresh_row, from_=0.0, to=1.0, orient=tk.HORIZONTAL,
|
||
style="Horizontal.TScale",
|
||
command=lambda v: local_thresh.set(round(float(v), 2)))
|
||
scale.set(float(local_thresh.get()))
|
||
scale.pack(side=tk.LEFT, fill=tk.X, expand=True, padx=10)
|
||
self.thresh_scale = scale
|
||
|
||
def _apply_thresh():
|
||
try:
|
||
v = min(1.0, max(0.0, float(local_thresh.get())))
|
||
except Exception:
|
||
return
|
||
self.manager.set_threshold(v)
|
||
self.thresh_val.set(v)
|
||
self._set_result_text(f"识别阈值已更新为 {v:.2f}", "ok")
|
||
self._toast(f"阈值已更新为 {v:.2f}", "ok")
|
||
|
||
PillButton(thresh_row, text="应用", variant="secondary", size="sm",
|
||
command=_apply_thresh).pack(side=tk.LEFT)
|
||
|
||
Divider(tab_cfg, bg=COLORS["bg"], padx=18, pady=14).pack(fill=tk.X)
|
||
|
||
tk.Label(tab_cfg, text="上报方式", bg=COLORS["bg"], fg=COLORS["text"],
|
||
font=FONTS["h2"]).pack(anchor=tk.W, padx=18)
|
||
tk.Label(tab_cfg, text="模拟模式把图片复制到本地对象存储 media_store，JSON 里填 URL；真实模式先 POST 图片再上报结果。",
|
||
bg=COLORS["bg"], fg=COLORS["muted"], font=FONTS["small"], wraplength=700,
|
||
justify=tk.LEFT).pack(anchor=tk.W, padx=18)
|
||
|
||
api_row = tk.Frame(tab_cfg, bg=COLORS["bg"])
|
||
api_row.pack(fill=tk.X, padx=18, pady=(8, 0))
|
||
mode_var = tk.StringVar(value="mock" if self.result_api.mock else "real")
|
||
url_var = tk.StringVar(value=self.result_api.api_url)
|
||
|
||
def _sync_mode():
|
||
url_entry.config(state=tk.NORMAL if mode_var.get() == "real" else tk.DISABLED)
|
||
|
||
ttk.Radiobutton(api_row, text="模拟模式", value="mock", variable=mode_var,
|
||
command=_sync_mode).pack(side=tk.LEFT)
|
||
ttk.Radiobutton(api_row, text="真实接口", value="real", variable=mode_var,
|
||
command=_sync_mode).pack(side=tk.LEFT, padx=(12, 0))
|
||
|
||
url_row = tk.Frame(tab_cfg, bg=COLORS["bg"])
|
||
url_row.pack(fill=tk.X, padx=18, pady=(6, 0))
|
||
tk.Label(url_row, text="接口地址", bg=COLORS["bg"], fg=COLORS["text_body"],
|
||
font=FONTS["ui"]).pack(side=tk.LEFT)
|
||
url_entry = ttk.Entry(url_row, textvariable=url_var, style="Large.TEntry")
|
||
url_entry.pack(side=tk.LEFT, fill=tk.X, expand=True, padx=8)
|
||
|
||
base_row = tk.Frame(tab_cfg, bg=COLORS["bg"])
|
||
base_row.pack(fill=tk.X, padx=18, pady=(8, 0))
|
||
tk.Label(base_row, text="基地编码", bg=COLORS["bg"], fg=COLORS["text_body"],
|
||
font=FONTS["ui"]).pack(side=tk.LEFT)
|
||
base_var = tk.StringVar(value=self.result_api.base_id or "")
|
||
ttk.Entry(base_row, textvariable=base_var, style="Large.TEntry").pack(
|
||
side=tk.LEFT, fill=tk.X, expand=True, padx=8)
|
||
|
||
def _apply_api():
|
||
self.result_api.mock = (mode_var.get() == "mock")
|
||
url = (url_var.get() or "").strip()
|
||
if url:
|
||
self.result_api.api_url = url
|
||
self.result_api.media_upload_url = default_media_upload_url(url)
|
||
bid = (base_var.get() or "").strip()
|
||
if bid:
|
||
self.result_api.base_id = bid
|
||
if hasattr(self, "api_badge"):
|
||
self.api_badge.set("API模拟" if self.result_api.mock else "API在线",
|
||
"warn" if self.result_api.mock else "ok")
|
||
tip = "已切换为模拟上报" if self.result_api.mock else f"已切换为真实接口：{self.result_api.api_url}"
|
||
self._set_result_text(tip, "ok")
|
||
self._toast(tip, "ok")
|
||
|
||
_sync_mode()
|
||
|
||
sync_btns = tk.Frame(tab_cfg, bg=COLORS["bg"])
|
||
sync_btns.pack(fill=tk.X, padx=18, pady=(16, 0))
|
||
PillButton(sync_btns, text="立即同步", variant="secondary",
|
||
command=self.sync_templates_now).pack(side=tk.LEFT, padx=(0, 8))
|
||
PillButton(sync_btns, text="同步详情", variant="ghost",
|
||
command=self.show_sync_dialog).pack(side=tk.LEFT, padx=(0, 8))
|
||
PillButton(sync_btns, text="识别统计", variant="ghost",
|
||
command=self.show_statistics_dialog).pack(side=tk.LEFT)
|
||
|
||
footer = tk.Frame(tab_cfg, bg=COLORS["bg"])
|
||
footer.pack(fill=tk.X, padx=18, pady=(18, 14), side=tk.BOTTOM)
|
||
PillButton(footer, text="应用上报设置", variant="primary",
|
||
command=_apply_api).pack(side=tk.RIGHT, padx=(8, 0))
|
||
PillButton(footer, text="关闭", variant="ghost",
|
||
command=_close_settings).pack(side=tk.RIGHT)
|
||
PillButton(footer, text="打开 mock 目录", variant="ghost", size="sm",
|
||
command=lambda: self._open_path(_MOCK_DIR)).pack(side=tk.LEFT)
|
||
PillButton(footer, text="打开模板目录", variant="ghost", size="sm",
|
||
command=lambda: self._open_path(TEMPLATE_SAVE_DIR)).pack(side=tk.LEFT, padx=8)
|
||
|
||
def _open_path(self, path):
|
||
try:
|
||
os.makedirs(path, exist_ok=True)
|
||
os.startfile(path) # noqa: S606 - Windows 专用
|
||
except Exception as e:
|
||
messagebox.showwarning("提示", f"无法打开目录：{e}")
|
||
|
||
# ---------- 统计 ----------
|
||
def show_statistics_dialog(self):
|
||
"""识别统计：基于本地识别记录聚合。"""
|
||
dialog = self._dialog_shell("识别统计", 720, 540)
|
||
|
||
try:
|
||
records = self.recognition_repo.list_recent(limit=500)
|
||
except Exception as e:
|
||
messagebox.showerror("错误", f"读取识别记录失败：{e}")
|
||
return
|
||
|
||
total = len(records)
|
||
dish_total = sum(int(r.get("dish_count") or 0) for r in records)
|
||
unknown_total = sum(int(r.get("unknown_count") or 0) for r in records)
|
||
known_total = max(dish_total - unknown_total, 0)
|
||
hit_rate = (known_total / dish_total * 100) if dish_total else 0.0
|
||
|
||
tk.Label(dialog, text=f"基于最近 {total} 次识别记录聚合（最多取样 500 条）",
|
||
bg=COLORS["bg"], fg=COLORS["muted"], font=FONTS["small"],
|
||
).pack(anchor=tk.W, padx=18, pady=(14, 8))
|
||
|
||
tiles = tk.Frame(dialog, bg=COLORS["bg"])
|
||
tiles.pack(fill=tk.X, padx=18)
|
||
specs = [
|
||
("识别次数", total, "info"),
|
||
("菜品总数", dish_total, "info"),
|
||
("已知菜", known_total, "ok"),
|
||
("未知菜", unknown_total, "warn"),
|
||
("命中率", f"{hit_rate:.0f}%", "ok" if hit_rate >= 70 else "warn"),
|
||
]
|
||
for i, (label, value, tone) in enumerate(specs):
|
||
tile = StatTile(tiles, value=value, label=label, tone=tone, bg=COLORS["bg"], width=118)
|
||
tile.pack(side=tk.LEFT, padx=(0, 8))
|
||
|
||
card = RoundCard(dialog, title="菜品出现次数 Top 20", radius=RADIUS["lg"],
|
||
padding=10, expand_body=True)
|
||
card.pack(fill=tk.BOTH, expand=True, padx=12, pady=14)
|
||
|
||
counter = {}
|
||
for r in records:
|
||
try:
|
||
items = json.loads(r.get("result_json") or "[]")
|
||
except Exception:
|
||
items = []
|
||
for it in items:
|
||
name = it.get("name") or (it.get("suggested_name") and f"未知近似：{it.get('suggested_name')}")
|
||
if not name:
|
||
continue
|
||
counter[name] = counter.get(name, 0) + 1
|
||
|
||
if not counter:
|
||
EmptyState(card.body, text="暂无可用统计数据",
|
||
hint="完成若干次识别后，这里会显示菜品出现频次排名",
|
||
glyph="◔", bg=COLORS["surface"]).pack(fill=tk.BOTH, expand=True)
|
||
else:
|
||
wrap, inner, _cv = self._make_scroll_area(card.body, COLORS["surface"])
|
||
wrap.pack(fill=tk.BOTH, expand=True)
|
||
ranked = sorted(counter.items(), key=lambda kv: (-kv[1], kv[0]))[:20]
|
||
top = ranked[0][1] if ranked else 1
|
||
for rank, (name, cnt) in enumerate(ranked, 1):
|
||
row = tk.Frame(inner, bg=COLORS["surface"])
|
||
row.pack(fill=tk.X, pady=2)
|
||
tk.Label(row, text=f"{rank:>2}", bg=COLORS["surface"], fg=COLORS["muted"],
|
||
font=FONTS["mono"], width=3, anchor="e").pack(side=tk.LEFT)
|
||
tk.Label(row, text=name, bg=COLORS["surface"], fg=COLORS["text"],
|
||
font=FONTS["ui"], width=14, anchor="w").pack(side=tk.LEFT, padx=(6, 6))
|
||
bar_w = int(220 * (cnt / top)) if top else 0
|
||
bar = tk.Canvas(row, width=224, height=12, bg=COLORS["surface"],
|
||
highlightthickness=0, bd=0)
|
||
bar.pack(side=tk.LEFT)
|
||
bar.create_rectangle(0, 2, max(bar_w, 2), 10, fill=COLORS["primary"], outline="")
|
||
tk.Label(row, text=f"{cnt} 次", bg=COLORS["surface"], fg=COLORS["text_body"],
|
||
font=FONTS["small"]).pack(side=tk.LEFT, padx=8)
|
||
|
||
PillButton(dialog, text="关闭", variant="ghost",
|
||
command=dialog.destroy).pack(pady=(0, 12))
|
||
|
||
# ---------- 预览 / 标注 ----------
|
||
def _preview_target_size(self):
|
||
"""预览区可用尺寸（尽量铺满右侧面板）。"""
|
||
max_w, max_h = 960, 720
|
||
try:
|
||
host = getattr(self, "preview_host", None)
|
||
if host is not None:
|
||
host.update_idletasks()
|
||
hw = int(host.winfo_width() or 0)
|
||
hh = int(host.winfo_height() or 0)
|
||
if hw > 40 and hh > 40:
|
||
return max(hw - 4, 320), max(hh - 4, 240)
|
||
panel = getattr(self, "preview_panel", None)
|
||
if panel is not None:
|
||
panel.update_idletasks()
|
||
pw = int(panel.winfo_width() or 0)
|
||
ph = int(panel.winfo_height() or 0)
|
||
if pw > 40 and ph > 40:
|
||
return max(pw - 8, 320), max(ph - 8, 240)
|
||
except Exception:
|
||
pass
|
||
return max_w, max_h
|
||
|
||
def _scale_preview_image(self, im):
|
||
"""等比放大/缩小到铺满预览区（允许放大，避免画面偏小）。"""
|
||
max_w, max_h = self._preview_target_size()
|
||
iw, ih = im.size
|
||
if iw <= 0 or ih <= 0:
|
||
return im
|
||
scale = min(max_w / float(iw), max_h / float(ih))
|
||
nw = max(1, int(round(iw * scale)))
|
||
nh = max(1, int(round(ih * scale)))
|
||
if (nw, nh) == (iw, ih):
|
||
return im
|
||
try:
|
||
resample = Image.Resampling.LANCZOS
|
||
except AttributeError:
|
||
resample = Image.LANCZOS
|
||
return im.resize((nw, nh), resample)
|
||
|
||
def show_preview_image(self, image_path):
|
||
self.stop_loading()
|
||
with open(image_path, 'rb') as f:
|
||
im = Image.open(f).convert("RGB")
|
||
im = self._scale_preview_image(im)
|
||
photo = ImageTk.PhotoImage(im)
|
||
self.img_label.config(text="", image=photo, bg=COLORS["preview_mid"], fg=COLORS["preview_fg"])
|
||
self.img_label.image = photo
|
||
|
||
def _preview_placeholder(self):
|
||
return "🖼\n等待相机画面…\n连接后将自动实时扫描"
|
||
|
||
def start_loading(self, tip="正在识别图片..."):
|
||
self._loading = True
|
||
self._loading_frame = 0
|
||
self.img_label.config(image="", text="")
|
||
self.img_label.image = None
|
||
self._set_result_text(tip, "busy")
|
||
self._tick_loading()
|
||
|
||
def _tick_loading(self):
|
||
if not getattr(self, "_loading", False):
|
||
return
|
||
frames = ["◐", "◓", "◑", "◒"]
|
||
ch = frames[self._loading_frame % len(frames)]
|
||
self.img_label.config(
|
||
text=f"{ch} 正在识别，请稍候…",
|
||
image="",
|
||
bg=COLORS["preview_mid"],
|
||
fg=COLORS["preview_fg"],
|
||
font=FONTS["section"],
|
||
)
|
||
self._loading_frame += 1
|
||
self._loading_job = self.root.after(120, self._tick_loading)
|
||
|
||
def stop_loading(self):
|
||
self._loading = False
|
||
if getattr(self, "_loading_job", None):
|
||
try:
|
||
self.root.after_cancel(self._loading_job)
|
||
except Exception:
|
||
pass
|
||
self._loading_job = None
|
||
|
||
def build_dish_labeled_image(self, source_image_path, boxes, dish_labels, styles=None):
|
||
if isinstance(source_image_path, np.ndarray):
|
||
stem = f"frame_{int(time.time() * 1000)}"
|
||
image_src = source_image_path
|
||
else:
|
||
stem = Path(source_image_path).stem
|
||
image_src = source_image_path
|
||
save_path = os.path.join(self.cfg.OUTPUT_DIR, f"{stem}.jpg")
|
||
return self.app.visualize_with_labels(
|
||
image_src, boxes, dish_labels, save_path, styles=styles,
|
||
)
|
||
|
||
def _sync_thresh_from_scale(self, value):
|
||
try:
|
||
self.thresh_val.set(round(float(value), 2))
|
||
except Exception:
|
||
pass
|
||
|
||
def _sync_thresh_from_spin(self, *_args):
|
||
try:
|
||
v = float(self.thresh_val.get())
|
||
v = min(1.0, max(0.0, v))
|
||
if self._widget_alive(getattr(self, "thresh_scale", None)):
|
||
self.thresh_scale.set(v)
|
||
except Exception:
|
||
pass
|
||
|
||
# ===============================================================
|
||
# UI 构建
|
||
# ===============================================================
|
||
def _build_header(self, c):
|
||
header = tk.Frame(self.root, bg=c["header_bg"], height=SIZES["header_h"])
|
||
header.pack(fill=tk.X, side=tk.TOP)
|
||
header.pack_propagate(False)
|
||
|
||
left_h = tk.Frame(header, bg=c["header_bg"])
|
||
left_h.pack(side=tk.LEFT, padx=20, pady=8)
|
||
|
||
logo = tk.Canvas(left_h, width=36, height=36, bg=c["header_bg"],
|
||
highlightthickness=0, bd=0)
|
||
logo.pack(side=tk.LEFT, padx=(0, 10))
|
||
logo.create_oval(1, 1, 35, 35, fill=c["primary_soft"], outline="")
|
||
logo.create_text(18, 19, text=self.logo_glyph, fill=c["primary"], font=FONTS["h1"])
|
||
|
||
title_box = tk.Frame(left_h, bg=c["header_bg"])
|
||
title_box.pack(side=tk.LEFT)
|
||
tk.Label(
|
||
title_box, text=getattr(self, "app_title", "菜品识别系统"),
|
||
bg=c["header_bg"], fg=c["text"], font=FONTS["title"],
|
||
).pack(anchor=tk.W)
|
||
|
||
right_h = tk.Frame(header, bg=c["header_bg"])
|
||
right_h.pack(side=tk.RIGHT, padx=16)
|
||
|
||
self.camera_badge = StatusBadge(right_h, text=f"{self.camera_display_name}未连接",
|
||
tone="err", tooltip="相机连接状态")
|
||
self.camera_badge.pack(side=tk.LEFT, padx=4)
|
||
api_tone = "warn" if self.result_api.mock else "ok"
|
||
api_text = "API模拟" if self.result_api.mock else "API在线"
|
||
self.api_badge = StatusBadge(right_h, text=api_text, tone=api_tone,
|
||
tooltip="识别结果上报方式")
|
||
self.api_badge.pack(side=tk.LEFT, padx=4)
|
||
self.base_badge = StatusBadge(right_h, text=self._format_base_sync_status(), tone="mute",
|
||
tooltip="本机基地与总库同步状态")
|
||
self.base_badge.pack(side=tk.LEFT, padx=4)
|
||
try:
|
||
self.base_badge.bind("<Button-1>", lambda e: self.show_sync_dialog())
|
||
self.base_badge._canvas.bind("<Button-1>", lambda e: self.show_sync_dialog())
|
||
self.base_badge._canvas.configure(cursor="hand2")
|
||
except Exception:
|
||
pass
|
||
|
||
PillButton(right_h, text="设置", variant="secondary", size="sm",
|
||
command=self.show_settings_dialog,
|
||
tooltip="学菜 / 模板库 / 阈值 / 同步").pack(side=tk.LEFT, padx=(8, 0))
|
||
|
||
self.camera_status_label = self.camera_badge
|
||
self.api_status_label = self.api_badge
|
||
self.base_status_label = self.base_badge
|
||
self._syncing = False
|
||
self._sync_dialog = None
|
||
tk.Frame(self.root, bg=c["border"], height=1).pack(fill=tk.X)
|
||
|
||
def _build_pos_cart(self, parent, c):
|
||
panel = tk.Frame(parent, bg=c["surface"], highlightbackground=c["border"], highlightthickness=1)
|
||
panel.pack(fill=tk.BOTH, expand=True)
|
||
|
||
head = tk.Frame(panel, bg=c["surface"])
|
||
head.pack(fill=tk.X, padx=12, pady=(12, 8))
|
||
tk.Label(head, text="已点菜品", bg=c["surface"], fg=c["text"],
|
||
font=FONTS["h1"]).pack(side=tk.LEFT)
|
||
self.cart_count_label = tk.Label(head, text="0 项", bg=c["surface"], fg=c["muted"],
|
||
font=FONTS["small"])
|
||
self.cart_count_label.pack(side=tk.RIGHT)
|
||
|
||
cols = tk.Frame(panel, bg=c["pos_row_alt"])
|
||
cols.pack(fill=tk.X)
|
||
for text, w in (("菜品", 22), ("数量", 10), ("", 4)):
|
||
tk.Label(cols, text=text, bg=c["pos_row_alt"], fg=c["muted"],
|
||
font=FONTS["small"], width=w, anchor="w").pack(side=tk.LEFT, padx=4, pady=6)
|
||
|
||
wrap, self.cart_frame, self._cart_canvas = self._make_scroll_area(panel, c["surface"])
|
||
wrap.pack(fill=tk.BOTH, expand=True, padx=4, pady=4)
|
||
return panel
|
||
|
||
def _build_pos_preview(self, parent, c):
|
||
panel = tk.Frame(parent, bg=c["bg"])
|
||
panel.pack(fill=tk.BOTH, expand=True)
|
||
panel.rowconfigure(1, weight=1)
|
||
panel.columnconfigure(0, weight=1)
|
||
|
||
tip = tk.Frame(panel, bg=c["pos_tip_bar"], height=32)
|
||
tip.grid(row=0, column=0, sticky="ew")
|
||
tip.grid_propagate(False)
|
||
self.scan_tip_label = tk.Label(
|
||
tip, text="实时扫描中 · 未知不入清单，稳定约 5 秒后记入待学习",
|
||
bg=c["pos_tip_bar"], fg=c["pos_tip_fg"], font=FONTS["h2"],
|
||
)
|
||
self.scan_tip_label.pack(expand=True)
|
||
|
||
preview_wrap = tk.Frame(panel, bg=c["surface"], highlightbackground=c["border"],
|
||
highlightthickness=1)
|
||
preview_wrap.grid(row=1, column=0, sticky="nsew", pady=(4, 0))
|
||
self.preview_panel = RoundedPreview(preview_wrap, radius=RADIUS["md"])
|
||
self.preview_panel.pack(fill=tk.BOTH, expand=True, padx=2, pady=2)
|
||
self.preview_host = self.preview_panel.host
|
||
self.img_label = tk.Label(
|
||
self.preview_host,
|
||
text=self._preview_placeholder(),
|
||
bg=c["preview_mid"],
|
||
fg=c["preview_fg"],
|
||
font=FONTS["preview_hint"],
|
||
justify=tk.CENTER,
|
||
)
|
||
self.img_label.pack(fill=tk.BOTH, expand=True)
|
||
self._loading = False
|
||
self._loading_job = None
|
||
self._loading_frame = 0
|
||
return panel
|
||
|
||
def _build_pos_bottom(self, c):
|
||
bar = tk.Frame(self.root, bg=c["pos_bottom_bar"], height=SIZES.get("pos_bottom_h", 72))
|
||
bar.pack(side=tk.BOTTOM, fill=tk.X)
|
||
bar.pack_propagate(False)
|
||
tk.Frame(self.root, bg=c["border"], height=1).pack(side=tk.BOTTOM, fill=tk.X)
|
||
|
||
inner = tk.Frame(bar, bg=c["pos_bottom_bar"])
|
||
inner.pack(fill=tk.BOTH, expand=True, padx=16, pady=10)
|
||
|
||
self.camera_btn = PillButton(
|
||
inner, text="暂停扫描", variant="primary",
|
||
command=self.toggle_live_scan,
|
||
)
|
||
self.camera_btn.pack(side=tk.LEFT, padx=(0, 10))
|
||
PillButton(inner, text="选图识别", variant="primary",
|
||
command=self.choose_and_predict_img).pack(side=tk.LEFT, padx=(0, 10))
|
||
PillButton(inner, text="清空", variant="ghost",
|
||
command=self.clear_cart).pack(side=tk.LEFT, padx=(0, 10))
|
||
self.setup_pending_learning_ui(inner, side=tk.LEFT, fill=tk.NONE, expand=False,
|
||
padx=(0, 10), pady=0)
|
||
PillButton(inner, text="设置", variant="secondary",
|
||
command=self.show_settings_dialog).pack(side=tk.RIGHT)
|
||
# 相机状态同步需要该属性；真正学菜按钮在设置页创建
|
||
self.camera_add_btn = self.camera_btn
|
||
return bar
|
||
|
||
def _build_sidebar(self, parent, c):
|
||
"""模板库侧栏（设置页嵌入）。"""
|
||
left_card = tk.Frame(parent, bg=c["sidebar"])
|
||
left_card.pack(fill=tk.BOTH, expand=True, padx=12, pady=12)
|
||
|
||
head_row = tk.Frame(left_card, bg=c["sidebar"])
|
||
head_row.pack(fill=tk.X, pady=(0, 10))
|
||
SectionTitle(head_row, "模板库", bg=c["sidebar"]).pack(side=tk.LEFT)
|
||
IconButton(head_row, text="🗑", command=self.clear_lib,
|
||
tooltip="清空所有模板").pack(side=tk.RIGHT)
|
||
|
||
self.search_box = SearchBox(
|
||
left_card, placeholder="搜索菜名…", command=self._on_search_change,
|
||
bg=c["sidebar"], height=32,
|
||
)
|
||
self.search_box.pack(fill=tk.X, pady=(0, 8))
|
||
|
||
list_wrap, self.scrollable_frame, self._template_canvas = self._make_scroll_area(
|
||
left_card, c["sidebar"])
|
||
list_wrap.pack(fill=tk.BOTH, expand=True)
|
||
|
||
self.template_stats = tk.Label(
|
||
left_card, text="共 0 个菜品 · 0 张模板",
|
||
bg=c["sidebar"], fg=c["muted"], font=FONTS["small"], anchor="w",
|
||
)
|
||
self.template_stats.pack(fill=tk.X, pady=(8, 4))
|
||
|
||
bottom_ops = tk.Frame(left_card, bg=c["sidebar"])
|
||
bottom_ops.pack(fill=tk.X, pady=(4, 0))
|
||
self.setup_pending_learning_ui(bottom_ops, side=tk.LEFT, fill=tk.X, expand=True,
|
||
padx=(0, 4), pady=0)
|
||
PillButton(
|
||
bottom_ops, text="识别记录", variant="ghost",
|
||
command=self.show_recognition_history,
|
||
).pack(side=tk.LEFT, fill=tk.X, expand=True, padx=(4, 0))
|
||
|
||
sync_row = tk.Frame(left_card, bg=c["sidebar"])
|
||
sync_row.pack(fill=tk.X, pady=(6, 0))
|
||
PillButton(
|
||
sync_row, text="立即同步", variant="secondary", command=self.sync_templates_now,
|
||
).pack(side=tk.LEFT, fill=tk.X, expand=True, padx=(0, 4))
|
||
PillButton(
|
||
sync_row, text="同步详情", variant="ghost", command=self.show_sync_dialog,
|
||
).pack(side=tk.LEFT, fill=tk.X, expand=True, padx=(4, 0))
|
||
return left_card
|
||
|
||
def _build_learn_panel(self, parent, c):
|
||
"""学菜面板（设置页）。"""
|
||
wrap = tk.Frame(parent, bg=c["bg"])
|
||
wrap.pack(fill=tk.BOTH, expand=True, padx=12, pady=12)
|
||
card = RoundCard(wrap, title="模板学习", subtitle="输入菜名后添加模板，或拍照学习")
|
||
card.pack(fill=tk.X)
|
||
row = tk.Frame(card.body, bg=c["surface"])
|
||
row.pack(fill=tk.X)
|
||
tk.Label(row, text="菜名", bg=c["surface"], fg=c["text_body"],
|
||
font=FONTS["ui"]).pack(side=tk.LEFT)
|
||
ttk.Entry(row, textvariable=self.class_name, width=18,
|
||
style="Large.TEntry").pack(side=tk.LEFT, padx=8)
|
||
PillButton(row, text="添加模板图片", variant="secondary",
|
||
command=self.add_template).pack(side=tk.LEFT, padx=4)
|
||
self.camera_add_btn = PillButton(
|
||
row, text="拍照添加模板", variant="secondary",
|
||
command=self.camera_capture_and_add_template,
|
||
)
|
||
self.camera_add_btn.pack(side=tk.LEFT, padx=4)
|
||
if not self.camera_connected:
|
||
try:
|
||
self.camera_add_btn.config(state=tk.DISABLED)
|
||
except Exception:
|
||
pass
|
||
return wrap
|
||
|
||
def create_ui(self):
|
||
c = COLORS
|
||
self.root.configure(bg=c["bg"])
|
||
self._build_header(c)
|
||
self._build_pos_bottom(c)
|
||
|
||
self.notice_bar = NoticeBar(self.root)
|
||
self.notice_bar.pack(side=tk.BOTTOM, fill=tk.X, padx=10, pady=(0, 6))
|
||
self.result_label = self.notice_bar
|
||
|
||
body = tk.Frame(self.root, bg=c["bg"])
|
||
body.pack(fill=tk.BOTH, expand=True, padx=8, pady=6)
|
||
body.columnconfigure(0, weight=0, minsize=SIZES.get("cart_w", 320))
|
||
body.columnconfigure(1, weight=1)
|
||
body.rowconfigure(0, weight=1)
|
||
|
||
left = tk.Frame(body, bg=c["bg"], width=SIZES.get("cart_w", 320))
|
||
left.grid(row=0, column=0, sticky="nsew", padx=(0, 8))
|
||
left.grid_propagate(False)
|
||
self._build_pos_cart(left, c)
|
||
|
||
right = tk.Frame(body, bg=c["bg"])
|
||
right.grid(row=0, column=1, sticky="nsew")
|
||
self._build_pos_preview(right, c)
|
||
|
||
self._set_result_text("就绪，等待识别…", "ok")
|
||
self.render_cart()
|
||
try:
|
||
self.root.minsize(1280, 800)
|
||
except Exception:
|
||
pass
|
||
|
||
# ---------------------------------------------------------------
|
||
# 模板列表
|
||
# ---------------------------------------------------------------
|
||
def _on_search_change(self, text):
|
||
self._template_filter = (text or "").strip().lower()
|
||
self.refresh_template_list()
|
||
|
||
def refresh_template_list(self):
|
||
frame = getattr(self, "scrollable_frame", None)
|
||
if not self._widget_alive(frame):
|
||
return
|
||
c = COLORS
|
||
self._thumb_refs = []
|
||
try:
|
||
children = list(frame.winfo_children())
|
||
except tk.TclError:
|
||
return
|
||
for widget in children:
|
||
try:
|
||
widget.destroy()
|
||
except tk.TclError:
|
||
pass
|
||
try:
|
||
items = self.manager.get_template_list()
|
||
except Exception:
|
||
items = []
|
||
dish_n, tpl_n = 0, 0
|
||
visible = [(n, cnt, th) for n, cnt, th in items if cnt > 0]
|
||
|
||
keyword = self._template_filter
|
||
matched = [(n, cnt, th) for n, cnt, th in visible
|
||
if not keyword or keyword in (n or "").lower()]
|
||
|
||
if not matched:
|
||
if visible and keyword:
|
||
EmptyState(
|
||
self.scrollable_frame, text="没有匹配的菜品",
|
||
hint=f"未找到包含「{keyword}」的模板，试试其他关键词",
|
||
glyph="◌", bg=c["sidebar"], pady=22,
|
||
).pack(fill=tk.X)
|
||
else:
|
||
EmptyState(
|
||
self.scrollable_frame, text="暂无模板",
|
||
hint="在「模板学习」中输入菜名并添加模板图片",
|
||
glyph="◍", bg=c["sidebar"], pady=22,
|
||
).pack(fill=tk.X)
|
||
stats = getattr(self, "template_stats", None)
|
||
if self._widget_alive(stats):
|
||
if keyword:
|
||
stats.config(text=f"筛选 0 / {len(visible)} 个菜品")
|
||
else:
|
||
stats.config(text="共 0 个菜品 · 0 张模板")
|
||
return
|
||
|
||
for name, count, _thumbnail in matched:
|
||
dish_n += 1
|
||
tpl_n += count
|
||
selected = self._selected_dish == name
|
||
border = c["row_selected"] if selected else c["border"]
|
||
bg0 = c["primary_soft"] if selected else c["surface"]
|
||
|
||
row = tk.Frame(
|
||
self.scrollable_frame, bg=bg0, cursor="hand2",
|
||
highlightbackground=border, highlightthickness=2 if selected else 1,
|
||
)
|
||
row.pack(fill=tk.X, pady=3, padx=2)
|
||
inner = tk.Frame(row, bg=bg0)
|
||
inner.pack(fill=tk.X, padx=10, pady=8)
|
||
|
||
# 42×42 圆角：优先真实模板图，否则渐变兜底
|
||
thumb = None
|
||
try:
|
||
details = self.manager.get_template_details(name)
|
||
if details:
|
||
thumb = make_photo_rounded(details[0].get("path"), size=42, radius=RADIUS["md"])
|
||
except Exception:
|
||
thumb = None
|
||
if thumb is None and _thumbnail is not None:
|
||
thumb = _thumbnail
|
||
if thumb is None:
|
||
thumb = make_rounded_thumb(name, size=42)
|
||
self._thumb_refs.append(thumb)
|
||
|
||
thumb_label = tk.Label(inner, image=thumb, bg=bg0, bd=0)
|
||
thumb_label.image = thumb
|
||
thumb_label.pack(side=tk.LEFT, padx=(0, 10))
|
||
|
||
text_frame = tk.Frame(inner, bg=bg0)
|
||
text_frame.pack(side=tk.LEFT, fill=tk.X, expand=True)
|
||
name_lbl = tk.Label(text_frame, text=name, bg=bg0, fg=c["text"], font=FONTS["ui_bold"])
|
||
name_lbl.pack(anchor=tk.W)
|
||
meta_lbl = tk.Label(
|
||
text_frame, text=f"{count} 张模板", bg=bg0, fg=c["muted"], font=FONTS["small"],
|
||
)
|
||
meta_lbl.pack(anchor=tk.W)
|
||
|
||
del_btn = ttk.Button(
|
||
inner, text="删除", style="GhostDanger.TButton",
|
||
command=lambda n=name: self.delete_template_category(n),
|
||
)
|
||
del_btn.pack(side=tk.RIGHT)
|
||
del_btn.pack_forget()
|
||
|
||
widgets = [row, inner, text_frame, thumb_label, name_lbl, meta_lbl]
|
||
|
||
def _paint(bg, widgets=widgets, border_c=border, thickness=1):
|
||
row.configure(bg=bg, highlightbackground=border_c, highlightthickness=thickness)
|
||
for w in widgets[1:]:
|
||
try:
|
||
w.configure(bg=bg)
|
||
except Exception:
|
||
pass
|
||
|
||
def _on_enter(_e=None, btn=del_btn, sel=selected):
|
||
if not sel:
|
||
_paint(c["row_hover"], border_c=c["primary"], thickness=1)
|
||
if not btn.winfo_ismapped():
|
||
btn.pack(side=tk.RIGHT)
|
||
|
||
def _on_leave(_e=None, btn=del_btn, sel=selected, box=row):
|
||
try:
|
||
x, y = box.winfo_pointerxy()
|
||
wx, wy = box.winfo_rootx(), box.winfo_rooty()
|
||
if wx <= x <= wx + box.winfo_width() and wy <= y <= wy + box.winfo_height():
|
||
return
|
||
except Exception:
|
||
pass
|
||
btn.pack_forget()
|
||
if sel:
|
||
_paint(c["primary_soft"], border_c=c["row_selected"], thickness=2)
|
||
else:
|
||
_paint(c["surface"], border_c=c["border"], thickness=1)
|
||
|
||
def _select(_e=None, n=name):
|
||
changed = self._selected_dish != n
|
||
self._selected_dish = n
|
||
if changed:
|
||
self.refresh_template_list()
|
||
self.show_template_detail(n)
|
||
|
||
for w in widgets:
|
||
w.bind("<Enter>", _on_enter)
|
||
w.bind("<Leave>", _on_leave)
|
||
w.bind("<Button-1>", _select)
|
||
del_btn.bind("<Enter>", _on_enter)
|
||
del_btn.bind("<Leave>", _on_leave)
|
||
|
||
stats = getattr(self, "template_stats", None)
|
||
if self._widget_alive(stats):
|
||
if keyword:
|
||
stats.config(
|
||
text=f"筛选 {dish_n} / {len(visible)} 个菜品 · {tpl_n} 张模板")
|
||
else:
|
||
stats.config(text=f"共 {dish_n} 个菜品 · {tpl_n} 张模板")
|
||
|
||
# ---------------------------------------------------------------
|
||
# 实时视频流扫描（约 1fps）
|
||
# ---------------------------------------------------------------
|
||
def _reset_live_stability(self):
|
||
self._live_stable_sig = None
|
||
self._live_stable_since = None
|
||
self._live_stable_snapshot = None
|
||
# committed_sig 保留到场景变化时再清，避免同一场景反复写入
|
||
|
||
def _live_result_signature(self, dishes):
|
||
"""粗粒度签名：盘数 + 已知/未知名 + 量化框（忽略细微分数抖动）。"""
|
||
parts = []
|
||
for d in dishes or []:
|
||
box = d.get("box") or []
|
||
if isinstance(box, (list, tuple)) and len(box) >= 4:
|
||
qb = tuple(int(round(float(x) / 40.0)) * 40 for x in box[:4])
|
||
else:
|
||
qb = (0, 0, 0, 0)
|
||
if d.get("is_unknown"):
|
||
sug = (d.get("suggested_category") or "").strip()[:24]
|
||
parts.append(("U", sug, qb))
|
||
else:
|
||
name = (d.get("label") or d.get("name") or "").strip()[:24]
|
||
parts.append(("K", name, qb))
|
||
parts.sort(key=lambda x: (x[2], x[0], x[1]))
|
||
return (len(parts), tuple(parts))
|
||
|
||
def _update_live_stability(self, dishes):
|
||
"""跟踪结果稳定性；满 5 秒且有未知时返回 True（应写入待学习一次）。"""
|
||
now = time.time()
|
||
unknowns = [d for d in (dishes or []) if d.get("is_unknown")]
|
||
if not dishes:
|
||
self._reset_live_stability()
|
||
return False
|
||
|
||
sig = self._live_result_signature(dishes)
|
||
if sig != self._live_stable_sig:
|
||
self._live_stable_sig = sig
|
||
self._live_stable_since = now
|
||
self._live_stable_snapshot = list(dishes)
|
||
# 场景变了，允许新场景再次写入
|
||
if sig != self._live_stable_committed_sig:
|
||
pass
|
||
return False
|
||
|
||
self._live_stable_snapshot = list(dishes)
|
||
since = self._live_stable_since or now
|
||
hold = float(getattr(self, "_live_stable_seconds", 5.0) or 5.0)
|
||
if now - since < hold:
|
||
return False
|
||
if not unknowns:
|
||
return False
|
||
if self._live_stable_committed_sig == sig:
|
||
return False
|
||
return True
|
||
|
||
def _commit_live_unknowns_to_pending(self, dishes):
|
||
"""将当前稳定画面中的未知菜写入待学习（复用已算特征，pending 内部还有去重）。"""
|
||
unknowns = [d for d in (dishes or []) if d.get("is_unknown")]
|
||
if not unknowns:
|
||
return 0
|
||
written = 0
|
||
os.makedirs(self.cfg.Cutting, exist_ok=True)
|
||
for d in unknowns:
|
||
feat = d.get("feature")
|
||
crop = d.get("crop_image")
|
||
crop_arr = d.get("crop_array")
|
||
if (not crop or not os.path.isfile(str(crop))) and crop_arr is not None:
|
||
try:
|
||
stamp = int(time.time() * 1000)
|
||
crop = os.path.abspath(
|
||
os.path.join(self.cfg.Cutting, f"_pending_{stamp}_{d.get('index', 0)}.jpeg")
|
||
)
|
||
if cv2.imwrite(crop, crop_arr):
|
||
d["crop_image"] = crop
|
||
except Exception as exc:
|
||
print(f"[live] write pending crop failed: {exc}")
|
||
crop = None
|
||
if not crop or not os.path.isfile(str(crop)):
|
||
continue
|
||
try:
|
||
if feat is not None:
|
||
pid = self.manager.pending_store.add(
|
||
crop,
|
||
feat,
|
||
d.get("suggested_category"),
|
||
d.get("score"),
|
||
)
|
||
else:
|
||
detail = self.predict_crop(crop, auto_pending=True)
|
||
pid = detail.get("pending_id")
|
||
except Exception as exc:
|
||
print(f"[live] pending commit failed: {exc}")
|
||
continue
|
||
if pid:
|
||
d["pending_id"] = pid
|
||
written += 1
|
||
self._live_stable_committed_sig = self._live_stable_sig
|
||
return written
|
||
|
||
def toggle_live_scan(self):
|
||
if self._live_enabled:
|
||
self.stop_live_scan()
|
||
else:
|
||
if not self.camera_connected:
|
||
messagebox.showwarning("提示", f"{self.camera_display_name}未连接，无法扫描！")
|
||
return
|
||
self.start_live_scan()
|
||
|
||
def stop_live_scan(self):
|
||
self._live_enabled = False
|
||
job = getattr(self, "_live_job", None)
|
||
if job is not None:
|
||
try:
|
||
self.root.after_cancel(job)
|
||
except Exception:
|
||
pass
|
||
self._live_job = None
|
||
if hasattr(self, "camera_btn"):
|
||
try:
|
||
self.camera_btn.config(text="继续扫描")
|
||
except Exception:
|
||
pass
|
||
pending_n = 0
|
||
try:
|
||
pending_n = int(self.manager.get_pending_count() or 0)
|
||
except Exception:
|
||
pending_n = 0
|
||
if pending_n > 0:
|
||
self._set_result_text(
|
||
f"扫描已暂停｜待学习库有 {pending_n} 个未知菜，可点击「待学习库」命名学习",
|
||
"warn",
|
||
)
|
||
self._toast(f"有 {pending_n} 个未知菜待学习", "warn")
|
||
if hasattr(self, "scan_tip_label"):
|
||
try:
|
||
self.scan_tip_label.config(
|
||
text=f"扫描已暂停 · 待学习库 {pending_n} 个，点击底栏「待学习库」处理"
|
||
)
|
||
except Exception:
|
||
pass
|
||
else:
|
||
self._set_result_text("实时扫描已暂停", "info")
|
||
if hasattr(self, "scan_tip_label"):
|
||
try:
|
||
self.scan_tip_label.config(text="扫描已暂停 · 点击「继续扫描」恢复")
|
||
except Exception:
|
||
pass
|
||
|
||
def start_live_scan(self):
|
||
if not self.camera_connected:
|
||
return
|
||
self._live_enabled = True
|
||
self._recognize_busy = False
|
||
self._reset_live_stability()
|
||
self._live_stable_committed_sig = None
|
||
if hasattr(self, "camera_btn"):
|
||
try:
|
||
self.camera_btn.config(text="暂停扫描")
|
||
except Exception:
|
||
pass
|
||
if hasattr(self, "scan_tip_label"):
|
||
try:
|
||
self.scan_tip_label.config(
|
||
text="实时扫描中 · 未知不入清单，稳定约 5 秒后记入待学习"
|
||
)
|
||
except Exception:
|
||
pass
|
||
self._set_result_text("实时扫描已开启（未知菜稳定 5 秒后写入待学习）", "ok")
|
||
self._schedule_live_tick(immediate=True)
|
||
|
||
def _schedule_live_tick(self, immediate=False):
|
||
if not self._live_enabled:
|
||
return
|
||
delay = 30 if immediate else int(getattr(self, "_live_interval_ms", 1000))
|
||
try:
|
||
self._live_job = self.root.after(delay, self._live_tick)
|
||
except Exception:
|
||
self._live_job = None
|
||
|
||
def _live_tick(self):
|
||
self._live_job = None
|
||
if not self._live_enabled or not self.camera_connected:
|
||
return
|
||
if not self._recognize_busy:
|
||
self._recognize_busy = True
|
||
threading.Thread(target=self._live_recognize_worker, daemon=True).start()
|
||
self._schedule_live_tick()
|
||
|
||
def show_preview_frame(self, frame_bgr):
|
||
"""在预览区显示 BGR 帧。"""
|
||
if frame_bgr is None or not hasattr(self, "img_label"):
|
||
return
|
||
try:
|
||
rgb = cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2RGB)
|
||
im = Image.fromarray(rgb)
|
||
im = self._scale_preview_image(im)
|
||
photo = ImageTk.PhotoImage(im)
|
||
self.img_label.config(text="", image=photo, bg=COLORS["preview_mid"], fg=COLORS["preview_fg"])
|
||
self.img_label.image = photo
|
||
except Exception as e:
|
||
print(f"[live] preview failed: {e}")
|
||
|
||
def _live_recognize_worker(self):
|
||
def _mark(msg):
|
||
fn = getattr(sys, "_dish_crash_mark", None)
|
||
if callable(fn):
|
||
try:
|
||
fn(msg)
|
||
except Exception:
|
||
pass
|
||
|
||
try:
|
||
_mark("live: grab frame")
|
||
ret, save_path, frame = self.take_live_frame()
|
||
if ret != 0 or frame is None:
|
||
return
|
||
|
||
stem = Path(save_path).stem if save_path else f"_live_{int(time.time() * 1000)}"
|
||
# 内存直传 YOLO；裁剪落盘供购物车缩略图 / 待学习；跳过中间 Plate 可视化
|
||
self.root.after(0, lambda: self._set_result_text("扫描中：YOLO 检测餐盘…", "busy"))
|
||
_mark("live: yolo start")
|
||
visualize_save_path, crop_items, boxes = self.app.process_image(
|
||
frame,
|
||
stem=stem,
|
||
save_crops=True,
|
||
visualize=False,
|
||
)
|
||
_mark(f"live: yolo done crops={len(crop_items)}")
|
||
cropped_file_paths = [c.get("path") for c in crop_items]
|
||
crop_arrays = [c.get("image") for c in crop_items]
|
||
|
||
if not crop_items:
|
||
self._reset_live_stability()
|
||
def _no_plate(raw=save_path, fr=frame):
|
||
if raw and os.path.exists(str(raw)):
|
||
self.show_preview_image(raw)
|
||
elif fr is not None:
|
||
self.show_preview_frame(fr)
|
||
self._set_result_text("扫描中：未检测到餐盘", "warn")
|
||
if hasattr(self, "scan_tip_label"):
|
||
try:
|
||
self.scan_tip_label.config(text="扫描中 · 请将餐盘放入画面中央")
|
||
except Exception:
|
||
pass
|
||
self.root.after(0, _no_plate)
|
||
return
|
||
|
||
self.root.after(
|
||
0,
|
||
lambda n=len(crop_items): self._set_result_text(
|
||
f"扫描中：已检出 {n} 个餐盘，正在识别菜品…", "busy"
|
||
),
|
||
)
|
||
_mark("live: feature/match start")
|
||
final_result, dish_labels, unknown_count, labeled_path, dishes = self.run_plate_predict_batch(
|
||
frame if frame is not None else save_path,
|
||
cropped_file_paths,
|
||
boxes,
|
||
auto_pending=False,
|
||
crop_images=crop_arrays,
|
||
draw_labels=False, # 实时避免 PIL RGBA 叠图导致原生闪退
|
||
)
|
||
_mark(f"live: feature/match done dishes={len(dishes)}")
|
||
|
||
# 轻量 OpenCV 画框（主线程展示前先在 worker 生成）
|
||
preview = save_path
|
||
try:
|
||
_mark("live: cv2 overlay start")
|
||
overlay = frame.copy()
|
||
for i, d in enumerate(dishes):
|
||
box = d.get("box") or (boxes[i] if i < len(boxes) else None)
|
||
if not box:
|
||
continue
|
||
x1, y1, x2, y2 = map(int, box)
|
||
color = (0, 180, 0) if not d.get("is_unknown") else (0, 165, 255)
|
||
cv2.rectangle(overlay, (x1, y1), (x2, y2), color, 2)
|
||
name = d.get("label") or d.get("suggested_category") or f"#{i+1}"
|
||
score = d.get("score")
|
||
text = f"{name}" if score is None else f"{name} {score:.2f}"
|
||
cv2.putText(
|
||
overlay, text, (x1, max(20, y1 - 8)),
|
||
cv2.FONT_HERSHEY_SIMPLEX, 0.55, color, 2, cv2.LINE_AA,
|
||
)
|
||
if save_path:
|
||
cv2.imwrite(save_path, overlay)
|
||
preview = save_path
|
||
_mark("live: cv2 overlay done")
|
||
except Exception as e:
|
||
_mark(f"live: cv2 overlay fail {e}")
|
||
preview = labeled_path or visualize_save_path or save_path
|
||
known_dishes = [d for d in dishes if not d.get("is_unknown")]
|
||
# 实时：仅有已识别菜才上报；排餐比对不刷 pending
|
||
if known_dishes and save_path:
|
||
self.report_recognition(
|
||
save_path, known_dishes, preview, schedule_check=False,
|
||
)
|
||
|
||
# 结果差不多连续满 5 秒 → 未知菜写入待学习（同场景一次）
|
||
should_commit = self._update_live_stability(dishes)
|
||
pending_written = 0
|
||
if should_commit:
|
||
pending_written = self._commit_live_unknowns_to_pending(
|
||
self._live_stable_snapshot or dishes
|
||
)
|
||
|
||
hold = float(getattr(self, "_live_stable_seconds", 5.0) or 5.0)
|
||
stable_elapsed = 0.0
|
||
if self._live_stable_since:
|
||
stable_elapsed = max(0.0, time.time() - self._live_stable_since)
|
||
already_committed = (
|
||
self._live_stable_committed_sig is not None
|
||
and self._live_stable_committed_sig == self._live_stable_sig
|
||
)
|
||
|
||
def update_ui(
|
||
path=preview, txt=final_result, ds=dishes,
|
||
written=pending_written, elapsed=stable_elapsed,
|
||
committed=already_committed, hold_s=hold,
|
||
):
|
||
if path and os.path.exists(str(path)):
|
||
self.show_preview_image(path)
|
||
self.render_result_dock(ds, path)
|
||
try:
|
||
self.update_pending_btn()
|
||
except Exception:
|
||
pass
|
||
|
||
known = [d for d in ds if not d.get("is_unknown")]
|
||
unknown = [d for d in ds if d.get("is_unknown")]
|
||
|
||
if known:
|
||
top = max(known, key=lambda d: d.get("score") or 0)
|
||
name = top.get("label") or "菜品"
|
||
msg = f"扫描中｜已识别 {len(known)}｜{name}"
|
||
if unknown:
|
||
msg += f"｜未知 {len(unknown)}（不入清单）"
|
||
self._set_result_text(msg, "ok" if not unknown else "warn")
|
||
else:
|
||
self._set_result_text(
|
||
txt or f"扫描中｜检出 {len(ds)} 盘，尚无达到阈值的菜品",
|
||
"warn",
|
||
)
|
||
|
||
if written:
|
||
self._toast(f"画面已稳定，{written} 个未知菜已记入待学习", "warn")
|
||
self._set_result_text(
|
||
f"未知菜已记入待学习（{written}）｜左侧仅显示已识别",
|
||
"warn",
|
||
)
|
||
|
||
if hasattr(self, "scan_tip_label"):
|
||
try:
|
||
if written:
|
||
self.scan_tip_label.config(
|
||
text=f"已记入待学习 {written} 个 · 场景变化后重新计时"
|
||
)
|
||
elif unknown and committed:
|
||
self.scan_tip_label.config(
|
||
text="扫描中 · 本场景未知菜已记入待学习"
|
||
)
|
||
elif unknown:
|
||
left = max(0, int(round(hold_s - elapsed)))
|
||
self.scan_tip_label.config(
|
||
text=f"扫描中 · 未知不入清单，再稳定约 {left}s 写入待学习"
|
||
)
|
||
else:
|
||
self.scan_tip_label.config(
|
||
text="实时扫描中 · 仅已识别菜计入左侧清单"
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
self.root.after(0, update_ui)
|
||
_mark("live: ui scheduled")
|
||
except Exception as e:
|
||
_mark(f"live: EXCEPTION {e}")
|
||
print(f"[live] recognize failed: {e}")
|
||
self.root.after(0, lambda: self._set_result_text(f"扫描异常：{e}", "err"))
|
||
finally:
|
||
self._recognize_busy = False
|
||
_mark("live: worker end")
|
||
|
||
# ---------------------------------------------------------------
|
||
# 结算机购物车
|
||
# ---------------------------------------------------------------
|
||
def clear_cart(self):
|
||
self._cart = []
|
||
self._last_dishes = []
|
||
self.render_cart()
|
||
if not self._live_enabled and hasattr(self, "img_label"):
|
||
self.img_label.config(
|
||
text=self._preview_placeholder(), image="",
|
||
bg=COLORS["preview_mid"], fg=COLORS["preview_fg"],
|
||
)
|
||
self.img_label.image = None
|
||
self._set_result_text("已清空本次已点菜品", "ok")
|
||
|
||
def fill_cart_from_dishes(self, dishes):
|
||
"""左侧清单只放已识别菜；未知仅留在预览标注 / 待学习流程。"""
|
||
cart = []
|
||
for d in dishes or []:
|
||
if d.get("is_unknown"):
|
||
continue
|
||
name = d.get("label") or d.get("name") or "菜品"
|
||
cart.append({
|
||
"index": d.get("index"),
|
||
"name": name,
|
||
"label": d.get("label"),
|
||
"score": d.get("score"),
|
||
"is_unknown": False,
|
||
"qty": 1,
|
||
"crop_image": d.get("crop_image"),
|
||
"pending_id": d.get("pending_id"),
|
||
"suggested_category": d.get("suggested_category"),
|
||
})
|
||
self._cart = cart
|
||
self._last_dishes = dishes or []
|
||
|
||
def _change_qty(self, idx, delta):
|
||
if idx < 0 or idx >= len(self._cart):
|
||
return
|
||
item = self._cart[idx]
|
||
item["qty"] = max(1, int(item.get("qty") or 1) + delta)
|
||
self.render_cart()
|
||
|
||
def _remove_cart_item(self, idx):
|
||
if 0 <= idx < len(self._cart):
|
||
self._cart.pop(idx)
|
||
self.render_cart()
|
||
|
||
def render_cart(self):
|
||
c = COLORS
|
||
if not hasattr(self, "cart_frame"):
|
||
return
|
||
for w in self.cart_frame.winfo_children():
|
||
w.destroy()
|
||
self._cart_thumbs = []
|
||
|
||
if not self._cart:
|
||
EmptyState(
|
||
self.cart_frame, text="暂无已识别菜品",
|
||
hint="未知菜不入清单；画面稳定约 5 秒后记入待学习",
|
||
glyph="◍", bg=c["surface"], pady=28,
|
||
).pack(fill=tk.X)
|
||
else:
|
||
for i, item in enumerate(self._cart):
|
||
bg = c["surface"] if i % 2 == 0 else c["pos_row_alt"]
|
||
row = tk.Frame(self.cart_frame, bg=bg)
|
||
row.pack(fill=tk.X, pady=1)
|
||
pad = tk.Frame(row, bg=bg)
|
||
pad.pack(fill=tk.X, padx=8, pady=8)
|
||
|
||
thumb = None
|
||
crop = item.get("crop_image")
|
||
if crop and os.path.exists(str(crop)):
|
||
try:
|
||
thumb = make_photo_rounded(crop, size=42, radius=RADIUS["sm"])
|
||
except Exception:
|
||
thumb = None
|
||
if thumb is None:
|
||
thumb = make_rounded_thumb(item.get("name") or "?", size=42)
|
||
self._cart_thumbs.append(thumb)
|
||
tk.Label(pad, image=thumb, bg=bg, bd=0).pack(side=tk.LEFT, padx=(0, 8))
|
||
|
||
info = tk.Frame(pad, bg=bg)
|
||
info.pack(side=tk.LEFT, fill=tk.X, expand=True)
|
||
tk.Label(info, text=item.get("name") or "菜品", bg=bg, fg=c["text"],
|
||
font=FONTS["ui_bold"], anchor="w").pack(fill=tk.X)
|
||
score = item.get("score")
|
||
meta = f"#{item.get('index') or i + 1}"
|
||
if score is not None:
|
||
meta += f" · {score:.2f}"
|
||
if item.get("is_unknown"):
|
||
meta += " · 待确认"
|
||
tk.Label(info, text=meta, bg=bg, fg=c["muted"],
|
||
font=FONTS["small"], anchor="w").pack(fill=tk.X)
|
||
|
||
qty_box = tk.Frame(pad, bg=bg)
|
||
qty_box.pack(side=tk.LEFT, padx=6)
|
||
PillButton(qty_box, text="−", variant="ghost", size="sm",
|
||
command=lambda i=i: self._change_qty(i, -1)).pack(side=tk.LEFT)
|
||
tk.Label(qty_box, text=str(item.get("qty") or 1), bg=bg, fg=c["text"],
|
||
font=FONTS["ui_bold"], width=3).pack(side=tk.LEFT)
|
||
PillButton(qty_box, text="+", variant="ghost", size="sm",
|
||
command=lambda i=i: self._change_qty(i, 1)).pack(side=tk.LEFT)
|
||
PillButton(pad, text="删", variant="ghost", size="sm",
|
||
command=lambda i=i: self._remove_cart_item(i)).pack(side=tk.LEFT, padx=(4, 0))
|
||
|
||
if hasattr(self, "cart_count_label"):
|
||
self.cart_count_label.config(text=f"{len(self._cart)} 项")
|
||
|
||
def render_result_dock(self, dishes, labeled_path=None):
|
||
"""兼容旧调用：识别结果写入购物车。"""
|
||
self.fill_cart_from_dishes(dishes)
|
||
self.render_cart()
|
||
|
||
# ---------------------------------------------------------------
|
||
# 模板详情 / 删除
|
||
# ---------------------------------------------------------------
|
||
def show_template_detail(self, part_type):
|
||
template_details = self.manager.get_template_details(part_type)
|
||
if not template_details:
|
||
messagebox.showinfo("提示", "该类别暂无模板信息")
|
||
return
|
||
detail_win = self._dialog_shell(f"模板详情 - {part_type}", 1000, 780)
|
||
|
||
tk.Label(detail_win, text=f"{part_type} · 共 {len(template_details)} 张模板",
|
||
bg=COLORS["bg"], fg=COLORS["text"], font=FONTS["h1"],
|
||
).pack(pady=(14, 8), padx=16, anchor=tk.W)
|
||
|
||
wrap, inner, _cv = self._make_scroll_area(detail_win, COLORS["bg"])
|
||
wrap.pack(fill=tk.BOTH, expand=True, padx=14, pady=(0, 8))
|
||
|
||
for idx, temp in enumerate(template_details, 1):
|
||
card = RoundCard(inner, title=f"模板 {idx}", radius=RADIUS["lg"], padding=10)
|
||
card.pack(fill=tk.X, pady=6)
|
||
if card.header is not None:
|
||
PillButton(card.header, text="删除此模板", variant="danger", size="sm",
|
||
command=lambda p=part_type, i=idx - 1: self.delete_single_template(p, i),
|
||
).pack(side=tk.RIGHT)
|
||
tk.Label(card.body, text=f"添加时间：{temp['add_time']}",
|
||
bg=COLORS["surface"], fg=COLORS["text_body"], font=FONTS["ui"],
|
||
anchor="w").pack(fill=tk.X)
|
||
tk.Label(card.body, text=f"文件路径：{temp['path']}",
|
||
bg=COLORS["surface"], fg=COLORS["muted"], font=FONTS["small"],
|
||
anchor="w", wraplength=860, justify=tk.LEFT).pack(fill=tk.X, pady=(2, 6))
|
||
try:
|
||
with open(temp['path'], 'rb') as f:
|
||
img = Image.open(f).convert("RGB")
|
||
img.thumbnail((860, 520), Image.Resampling.LANCZOS)
|
||
photo = ImageTk.PhotoImage(img)
|
||
lbl = tk.Label(card.body, image=photo, bg=COLORS["surface"], bd=0)
|
||
lbl.image = photo
|
||
lbl.pack()
|
||
except Exception as e:
|
||
tk.Label(card.body, text=f"图片加载失败：{e}", bg=COLORS["surface"],
|
||
fg=COLORS["danger"], font=FONTS["small"]).pack(pady=4)
|
||
|
||
PillButton(detail_win, text="关闭", variant="ghost",
|
||
command=detail_win.destroy).pack(pady=(0, 12))
|
||
|
||
def center_window(self, window, width, height):
|
||
try:
|
||
window.update_idletasks()
|
||
sw = window.winfo_screenwidth()
|
||
sh = window.winfo_screenheight()
|
||
width = min(width, max(sw - 80, 400))
|
||
height = min(height, max(sh - 100, 300))
|
||
x = (sw - width) // 2
|
||
y = max((sh - height) // 2 - 20, 20)
|
||
window.geometry(f"{width}x{height}+{x}+{y}")
|
||
except Exception:
|
||
window.geometry(f"{width}x{height}")
|
||
|
||
def delete_template_category(self, part_type):
|
||
if messagebox.askyesno("确认删除", f"确定要删除类别「{part_type}」的所有模板吗？"):
|
||
def delete_async():
|
||
success = self.manager.delete_template_category(part_type)
|
||
self.root.after(0, lambda: self._handle_delete_result(success, part_type, True))
|
||
threading.Thread(target=delete_async, daemon=True).start()
|
||
|
||
def delete_single_template(self, part_type, template_index):
|
||
if messagebox.askyesno("确认删除", f"确定要删除「{part_type}」的第{template_index + 1}个模板吗？"):
|
||
def delete_async():
|
||
success = self.manager.delete_template(part_type, template_index)
|
||
self.root.after(0, lambda: self._handle_delete_result(success, part_type, False))
|
||
threading.Thread(target=delete_async, daemon=True).start()
|
||
|
||
def _handle_delete_result(self, success, part_type, is_category):
|
||
if success:
|
||
self._toast(f"已删除类别「{part_type}」的全部模板" if is_category else "模板删除成功", "ok")
|
||
self.refresh_template_list()
|
||
for widget in self.root.winfo_children():
|
||
if isinstance(widget, tk.Toplevel) and "模板详情" in widget.title():
|
||
widget.destroy()
|
||
else:
|
||
messagebox.showerror("错误", "删除失败")
|
||
|
||
# ---------------------------------------------------------------
|
||
# 状态文案
|
||
# ---------------------------------------------------------------
|
||
def _set_result_text(self, text, tone="ok"):
|
||
"""tone: ok / warn / err / busy"""
|
||
if tone in ("muted", "info", "mute"):
|
||
tone = "ok"
|
||
raw = str(text)
|
||
# 已带语义前缀则不再叠加
|
||
has_mark = raw.startswith(("✅", "⚠️", "❌", "🟢", "🔴", "🔄", "◐", "◓", "◑", "◒"))
|
||
if not has_mark:
|
||
if tone == "ok":
|
||
raw = f"🟢 {raw}"
|
||
elif tone == "warn":
|
||
raw = f"⚠️ {raw}"
|
||
elif tone == "err":
|
||
raw = f"🔴 {raw}"
|
||
elif tone == "busy":
|
||
raw = f"🔄{raw}"
|
||
if hasattr(self, "notice_bar"):
|
||
self.notice_bar.set(raw, tone=tone)
|
||
return
|
||
if hasattr(self, "result_label") and hasattr(self.result_label, "set"):
|
||
self.result_label.set(raw, tone=tone)
|
||
return
|
||
try:
|
||
self.result_label.config(text=raw)
|
||
except Exception:
|
||
pass
|
||
|
||
def _update_camera_ui_success(self, camera_name=None):
|
||
name = camera_name or self.camera_display_name
|
||
self._set_result_text(f"{name}已连接，开始实时扫描", "ok")
|
||
if hasattr(self, "camera_badge"):
|
||
self.camera_badge.set(f"{name}已连接", "ok")
|
||
if hasattr(self, "camera_btn"):
|
||
self.camera_btn.config(state=tk.NORMAL)
|
||
if hasattr(self, "camera_add_btn"):
|
||
try:
|
||
self.camera_add_btn.config(state=tk.NORMAL)
|
||
except Exception:
|
||
pass
|
||
self.start_live_scan()
|
||
|
||
def _update_camera_ui_fail(self):
|
||
name = self.camera_display_name
|
||
tip = "相机断开，请检查USB设备" if "USB" in name else f"{name}未连接，请检查设备"
|
||
self.stop_live_scan()
|
||
self._set_result_text(tip, "err")
|
||
if hasattr(self, "camera_badge"):
|
||
self.camera_badge.set(f"{name}未连接", "err")
|
||
if hasattr(self, "camera_btn"):
|
||
self.camera_btn.config(state=tk.DISABLED, text="继续扫描")
|
||
if hasattr(self, "camera_add_btn"):
|
||
try:
|
||
self.camera_add_btn.config(state=tk.DISABLED)
|
||
except Exception:
|
||
pass
|
||
|
||
# ---------------------------------------------------------------
|
||
# 业务动作
|
||
# ---------------------------------------------------------------
|
||
def add_template(self):
|
||
cls = self.class_name.get().strip()
|
||
if not cls:
|
||
messagebox.showwarning("提示", "请输入类别名称！")
|
||
return
|
||
paths = filedialog.askopenfilenames(
|
||
title="选择模板图片",
|
||
filetypes=[("图片", "*.jpg *.jpeg *.png *.bmp"), ("所有文件", "*.*")],
|
||
)
|
||
if not paths:
|
||
return
|
||
|
||
def work():
|
||
try:
|
||
added, total = self.manager.add_template(cls, paths)
|
||
self.root.after(0, self.refresh_template_list)
|
||
self.root.after(0, lambda: self._toast(f"「{cls}」新增 {added} 张，共 {total} 张", "ok"))
|
||
except Exception as e:
|
||
self.root.after(0, lambda: messagebox.showerror("错误", f"添加失败：{str(e)}"))
|
||
|
||
threading.Thread(target=work, daemon=True).start()
|
||
|
||
def clear_lib(self):
|
||
if messagebox.askyesno("确认", "确定清空所有模板？"):
|
||
self.manager.clear_all()
|
||
self.refresh_template_list()
|
||
self.update_pending_btn()
|
||
self._set_result_text("模板库已清空", "info")
|
||
|
||
def choose_and_predict_img(self):
|
||
file_path = filedialog.askopenfilename(
|
||
title="选择识别图片",
|
||
filetypes=[("图片", "*.jpg *.jpeg *.png *.bmp"), ("所有文件", "*.*")],
|
||
)
|
||
if not file_path:
|
||
return
|
||
resume_live = self._live_enabled
|
||
if resume_live:
|
||
self._live_enabled = False
|
||
job = getattr(self, "_live_job", None)
|
||
if job is not None:
|
||
try:
|
||
self.root.after_cancel(job)
|
||
except Exception:
|
||
pass
|
||
self._live_job = None
|
||
self.start_loading("正在识别图片...")
|
||
self.root.update_idletasks()
|
||
try:
|
||
visualize_save_path, crop_items, boxes = self.app.process_image(
|
||
file_path, save_crops=True, visualize=True,
|
||
)
|
||
cropped_file_paths = [c.get("path") for c in crop_items]
|
||
crop_arrays = [c.get("image") for c in crop_items]
|
||
if not crop_items:
|
||
self.stop_loading()
|
||
self.img_label.config(
|
||
text=self._preview_placeholder(),
|
||
image="", bg=COLORS["preview_mid"], fg=COLORS["preview_fg"],
|
||
)
|
||
self._set_result_text("未检测到餐盘区域", "warn")
|
||
self._toast("未检测到餐盘区域", "warn")
|
||
return
|
||
|
||
final_result, _, _unknown_count, labeled_path, dishes = self.run_plate_predict_batch(
|
||
file_path, cropped_file_paths, boxes,
|
||
progress_callback=lambda cur, _total: self._set_result_text(
|
||
f"正在识别第 {cur} 张…", "busy"
|
||
),
|
||
crop_images=crop_arrays,
|
||
)
|
||
preview = labeled_path or visualize_save_path
|
||
if preview:
|
||
self.show_preview_image(preview)
|
||
else:
|
||
self.stop_loading()
|
||
self.render_result_dock(dishes, preview)
|
||
self.report_recognition(file_path, dishes, preview)
|
||
self._announce_result(dishes, final_result)
|
||
self.confirm_unknown_dishes(
|
||
dishes,
|
||
on_finished=lambda learned: learned and self.report_recognition(file_path, dishes, preview),
|
||
)
|
||
except Exception as e:
|
||
self.stop_loading()
|
||
self._set_result_text(f"识别异常：{e}", "err")
|
||
messagebox.showerror("错误", str(e))
|
||
finally:
|
||
if resume_live and self.camera_connected:
|
||
self.start_live_scan()
|
||
|
||
def _announce_result(self, dishes, fallback_text=""):
|
||
"""统一的结果播报：状态条摘要 + 轻提示。"""
|
||
known = [d for d in dishes if not d.get("is_unknown")]
|
||
unknown = [d for d in dishes if d.get("is_unknown")]
|
||
if known:
|
||
top = max(known, key=lambda d: d.get("score") or 0)
|
||
name = top.get("label") or "菜品"
|
||
score = top.get("score")
|
||
score_txt = f"{score:.2f}" if score is not None else "—"
|
||
msg = f"识别完成｜共{len(dishes)}个区域｜菜品：{name}｜置信度：{score_txt}"
|
||
if unknown:
|
||
msg += f"｜待确认 {len(unknown)}"
|
||
self._set_result_text(msg, "ok")
|
||
self._toast(msg, "ok" if not unknown else "warn")
|
||
else:
|
||
msg = fallback_text or f"识别完成｜共{len(dishes)}个区域"
|
||
self._set_result_text(msg, "warn")
|
||
self._toast("识别完成，但全部为未知菜品，请确认学习", "warn")
|
||
|
||
def camera_capture_and_add_template(self):
|
||
cls = self.class_name.get().strip()
|
||
if not cls:
|
||
messagebox.showwarning("提示", "请先输入类别名称！")
|
||
return
|
||
if not self.camera_connected:
|
||
messagebox.showwarning("提示", f"{self.camera_display_name}未连接，无法拍照！")
|
||
return
|
||
self.camera_add_btn.config(state=tk.DISABLED)
|
||
self._set_result_text("正在拍照添加模板...", "warn")
|
||
|
||
def camera_work():
|
||
try:
|
||
ret, save_path = self.take_photo()
|
||
if ret != 0 or not save_path or not isinstance(save_path, str) or not os.path.exists(str(save_path)):
|
||
self.root.after(0, lambda: messagebox.showerror("错误", "拍照失败！"))
|
||
return
|
||
_, cropped_file_paths, _ = self.app.process_single_image(save_path)
|
||
if cropped_file_paths:
|
||
added, total = self.manager.add_template(cls, cropped_file_paths)
|
||
if added > 0:
|
||
with open(cropped_file_paths[0], 'rb') as f:
|
||
im = Image.open(f).convert("RGB")
|
||
im.thumbnail((800, 600))
|
||
photo = ImageTk.PhotoImage(im)
|
||
self.root.after(0, lambda: self.img_label.config(text="", image=photo))
|
||
self.img_label.image = photo
|
||
self.root.after(0, self.refresh_template_list)
|
||
self.root.after(0, lambda: self._set_result_text(
|
||
f"模板添加成功！已添加{added}/{total}张", "ok"))
|
||
self.root.after(0, lambda: self._toast(
|
||
f"已添加为「{cls}」的模板（{added} 张）", "ok"))
|
||
else:
|
||
self.root.after(0, lambda: messagebox.showerror("错误", "模板添加失败"))
|
||
else:
|
||
self.root.after(0, lambda: messagebox.showerror("错误", "拍照失败或未检测到餐盘！"))
|
||
except Exception as e:
|
||
self.root.after(0, lambda: messagebox.showerror("错误", f"拍照失败：{str(e)}"))
|
||
finally:
|
||
self.root.after(0, lambda: self.camera_add_btn.config(state=tk.NORMAL))
|
||
|
||
threading.Thread(target=camera_work, daemon=True).start()
|
||
|
||
def camera_capture_and_predict(self):
|
||
if not self.camera_connected:
|
||
messagebox.showwarning("提示", f"{self.camera_display_name}未连接，无法拍照！")
|
||
return
|
||
self.camera_btn.config(state=tk.DISABLED)
|
||
self.start_loading("正在拍照识别...")
|
||
|
||
def _fail_preview(msg, tone="warn"):
|
||
self.stop_loading()
|
||
self.img_label.config(
|
||
text=self._preview_placeholder(),
|
||
image="", bg=COLORS["preview_mid"], fg=COLORS["preview_fg"],
|
||
)
|
||
self.img_label.image = None
|
||
self._set_result_text(msg, tone)
|
||
|
||
def camera_work():
|
||
try:
|
||
ret, save_path = self.take_photo()
|
||
if ret != 0 or not save_path or not isinstance(save_path, str) or not os.path.exists(str(save_path)):
|
||
self.root.after(0, lambda: messagebox.showerror("错误", "拍照失败！"))
|
||
self.root.after(0, lambda: _fail_preview("拍照失败", "err"))
|
||
return
|
||
|
||
self.root.after(0, lambda: self._set_result_text("正在识别图片...", "busy"))
|
||
visualize_save_path, crop_items, boxes = self.app.process_image(
|
||
save_path, save_crops=True, visualize=True,
|
||
)
|
||
cropped_file_paths = [c.get("path") for c in crop_items]
|
||
crop_arrays = [c.get("image") for c in crop_items]
|
||
if not crop_items:
|
||
self.root.after(0, lambda: messagebox.showwarning("提示", "未检测到餐盘区域"))
|
||
self.root.after(0, lambda: _fail_preview("未检测到餐盘区域", "warn"))
|
||
return
|
||
|
||
final_result, _, _unknown_count, labeled_path, dishes = self.run_plate_predict_batch(
|
||
save_path, cropped_file_paths, boxes, crop_images=crop_arrays,
|
||
)
|
||
preview = labeled_path or visualize_save_path
|
||
self.report_recognition(save_path, dishes, preview)
|
||
|
||
def update_ui(path=preview, txt=final_result, ds=dishes, src=save_path):
|
||
if path:
|
||
self.show_preview_image(path)
|
||
else:
|
||
self.stop_loading()
|
||
self.render_result_dock(ds, path)
|
||
self._announce_result(ds, txt)
|
||
unknowns = [d for d in ds if d.get("is_unknown") and d.get("pending_id")]
|
||
if unknowns:
|
||
self.confirm_unknown_dishes(
|
||
ds,
|
||
on_finished=lambda learned: learned and self.report_recognition(src, ds, path),
|
||
)
|
||
|
||
self.root.after(0, update_ui)
|
||
except Exception as e:
|
||
self.root.after(0, lambda: messagebox.showerror("错误", f"拍照/识别异常：{str(e)}"))
|
||
self.root.after(0, lambda: _fail_preview(f"识别异常：{e}", "err"))
|
||
finally:
|
||
self.root.after(0, lambda: self.camera_btn.config(state=tk.NORMAL))
|
||
|
||
threading.Thread(target=camera_work, daemon=True).start()
|
||
|
||
|
||
def run_app(app_cls):
|
||
"""统一启动入口（USB / 海康共用）。启动时一并拉起本机媒体服务。"""
|
||
import ctypes
|
||
import platform
|
||
|
||
print(platform.architecture())
|
||
if os.name == 'nt':
|
||
import multiprocessing
|
||
multiprocessing.freeze_support()
|
||
try:
|
||
ctypes.windll.shcore.SetProcessDpiAwareness(1)
|
||
except Exception:
|
||
pass
|
||
|
||
# 比 UI/YOLO 更早绑定 :9100，保证启动后即可访问裁剪图
|
||
try:
|
||
from service.media_server import ensure_media_server
|
||
ensure_media_server()
|
||
except Exception as exc:
|
||
print(f'[WARN] media server early start failed: {exc}')
|
||
|
||
root = tk.Tk()
|
||
app = app_cls(root)
|
||
|
||
def on_closing():
|
||
try:
|
||
app.stop_live_scan()
|
||
except Exception:
|
||
pass
|
||
try:
|
||
app.shutdown_mgmt_services()
|
||
except Exception:
|
||
pass
|
||
app.manager.save_templates()
|
||
app.close_camera()
|
||
root.destroy()
|
||
|
||
root.protocol("WM_DELETE_WINDOW", on_closing)
|
||
root.mainloop()
|