import os import json import shutil import datetime from collections import defaultdict from pathlib import Path import threading import cv2 import torch import torch.nn as nn import torchvision.models as models from torchvision import transforms from torchvision.models.convnext import ConvNeXt_Tiny_Weights from PIL import Image, ImageTk import numpy as np import tkinter as tk from tkinter import ttk, filedialog, messagebox from core.accel import resolve_feature_openvino, torch_device from core.pending_learning import PendingLearningStore from ui.pending_learning_ui import PendingLearningUIMixin, format_dish_box_label, format_predict_detail # -------------------------- # 设备配置 # -------------------------- device = torch_device() # 模板保存路径配置 TEMPLATE_SAVE_DIR = os.path.join(os.path.expanduser("~"), "ImageClassifierTemplates") TEMPLATE_DATA_FILE = os.path.join(TEMPLATE_SAVE_DIR, "templates_data.json") THUMBNAIL_DIR = os.path.join(TEMPLATE_SAVE_DIR, "thumbnails") # 创建保存目录 os.makedirs(TEMPLATE_SAVE_DIR, exist_ok=True) os.makedirs(THUMBNAIL_DIR, exist_ok=True) _IMAGENET_MEAN = np.array([0.485, 0.456, 0.406], dtype=np.float32) _IMAGENET_STD = np.array([0.229, 0.224, 0.225], dtype=np.float32) # -------------------------- # 全局字体配置 # -------------------------- def setup_ttk_styles(root=None): """配置 ttk 主题（委托 ui.theme）。""" from ui.theme import apply_theme return apply_theme(root) # -------------------------- # 特征提取模型 # -------------------------- class FeatureExtractor(nn.Module): def __init__(self): super().__init__() self.convnext = models.convnext_tiny(weights=ConvNeXt_Tiny_Weights.DEFAULT) self.features = nn.Sequential(*list(self.convnext.children())[:-1]) for param in self.parameters(): param.requires_grad = False self.feature_dim = 768 def forward(self, x): features = self.features(x) features = features.view(features.size(0), -1) features = nn.functional.normalize(features, p=2, dim=1) return features class OpenVINOFeatureBackend: """可选：用 OpenVINO 跑 ConvNeXt 特征（Intel CPU/iGPU）。""" def __init__(self, model_dir): from pathlib import Path from openvino import Core xmls = sorted(Path(model_dir).glob("*.xml")) if not xmls: raise FileNotFoundError(f"OpenVINO 模型目录无 xml: {model_dir}") core = Core() model = core.read_model(str(xmls[0])) # AUTO 会优先用 Intel GPU（若可用），否则 CPU self.compiled = core.compile_model(model, os.environ.get("DISH_OPENVINO_DEVICE", "AUTO")) self.output_key = self.compiled.outputs[0] self.feature_dim = 768 print(f"[OK] feature OpenVINO loaded: {xmls[0]} device={os.environ.get('DISH_OPENVINO_DEVICE', 'AUTO')}") def __call__(self, batch_nchw: np.ndarray) -> np.ndarray: out = self.compiled([batch_nchw])[self.output_key] feats = np.asarray(out, dtype=np.float32) if feats.ndim > 2: feats = feats.reshape(feats.shape[0], -1) norms = np.linalg.norm(feats, axis=1, keepdims=True) norms = np.maximum(norms, 1e-12) return feats / norms # -------------------------- # 模板管理器 # -------------------------- class PartTemplateManager: def __init__(self, similarity_threshold=0.7): self.transform = transforms.Compose([ transforms.Resize(256), transforms.CenterCrop(224), transforms.ToTensor(), transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]) ]) self._ov_backend = None ov_dir = resolve_feature_openvino() if ov_dir is not None: try: self._ov_backend = OpenVINOFeatureBackend(ov_dir) self.extractor = None self.feature_dim = 768 except Exception as exc: print(f"[WARN] OpenVINO feature backend failed, fallback torch: {exc}") self._ov_backend = None if self._ov_backend is None: self.extractor = FeatureExtractor().to(device).eval() self.feature_dim = self.extractor.feature_dim print(f"[OK] feature extractor on {device}") self.template_library = defaultdict(list) self.threshold = similarity_threshold self.lock = threading.Lock() # 矩阵相似度缓存：categories[i] 对应 matrix[i] self._sim_matrix = np.zeros((0, 768), dtype=np.float32) self._sim_categories = [] self.pending_store = PendingLearningStore(TEMPLATE_SAVE_DIR) from dao.template_repository import TemplateRepository self.repo = TemplateRepository(json_path=TEMPLATE_DATA_FILE) self.repo.set_reload_callback(self.load_templates) print("[INFO] loading templates into memory…") self.load_templates() print("[OK] PartTemplateManager ready") def get_base_info(self): return self.repo.get_base_info() def sync_now(self, full=True): return self.repo.sync_now(full=full) def get_sync_status(self): return self.repo.get_sync_status() def list_upload_queue(self, statuses=('pending', 'failed'), limit=100): return self.repo.list_upload_queue(statuses=statuses, limit=limit) def retry_failed_uploads(self): return self.repo.retry_failed_uploads() @property def source_base_id(self): return self.repo.source_base_id @property def base_code(self): return getattr(self.repo, 'base_code', None) def _rebuild_sim_index(self): """将模板特征压成 float32 矩阵，供一次 matmul 完成相似度。""" cats = [] rows = [] for cls_name, templates in self.template_library.items(): for temp in templates: feat = temp.get("feature") if feat is None: continue vec = np.asarray(feat, dtype=np.float32).reshape(-1) if vec.size == 0: continue rows.append(vec) cats.append(cls_name) if rows: mat = np.stack(rows, axis=0) # 兜底 L2，避免旧数据未归一化 norms = np.linalg.norm(mat, axis=1, keepdims=True) norms = np.maximum(norms, 1e-12) mat = mat / norms self._sim_matrix = mat self._sim_categories = cats else: self._sim_matrix = np.zeros((0, self.feature_dim), dtype=np.float32) self._sim_categories = [] @staticmethod def _to_pil_rgb(image_like): """路径 / BGR ndarray / RGB ndarray / PIL → RGB PIL。""" if image_like is None: return None if isinstance(image_like, Image.Image): return image_like.convert("RGB") if isinstance(image_like, np.ndarray): arr = image_like if arr.ndim == 2: arr = cv2.cvtColor(arr, cv2.COLOR_GRAY2RGB) elif arr.ndim == 3 and arr.shape[2] == 4: arr = cv2.cvtColor(arr, cv2.COLOR_BGRA2RGB) elif arr.ndim == 3 and arr.shape[2] == 3: # 热路径裁剪来自 OpenCV，默认按 BGR 处理 arr = cv2.cvtColor(arr, cv2.COLOR_BGR2RGB) else: return None return Image.fromarray(arr) if isinstance(image_like, (tuple, list)): image_like = image_like[0] if isinstance(image_like, str): try: image_like = image_like.encode("utf-8").decode("utf-8") except Exception: return None if not os.path.exists(image_like): return None try: with open(image_like, "rb") as f: return Image.open(f).convert("RGB") except Exception: return None return None def _preprocess_pil(self, img: Image.Image) -> np.ndarray: """返回 NCHW float32（单张），与 torchvision 变换一致。""" tensor = self.transform(img) # CHW return tensor.numpy().astype(np.float32) def _infer_features(self, batch_nchw: np.ndarray) -> np.ndarray: """batch: (N,3,224,224) float32 → (N,768) L2-normalized。""" if batch_nchw is None or len(batch_nchw) == 0: return np.zeros((0, self.feature_dim), dtype=np.float32) if self._ov_backend is not None: return self._ov_backend(batch_nchw) with torch.no_grad(): t = torch.from_numpy(batch_nchw).to(device) feats = self.extractor(t).cpu().numpy() return feats.astype(np.float32) def _extract_image_feature(self, image_like): img = self._to_pil_rgb(image_like) if img is None: return None try: nchw = self._preprocess_pil(img)[None, ...] feature = self._infer_features(nchw)[0] if np.isnan(feature).any() or np.isinf(feature).any(): return None return feature except Exception as e: print(f"特征提取失败: {e}") return None def extract_features_batch(self, image_likes): """批量提特征；失败位置为 None。""" results = [None] * len(image_likes) tensors = [] indices = [] for i, src in enumerate(image_likes): img = self._to_pil_rgb(src) if img is None: continue try: tensors.append(self._preprocess_pil(img)) indices.append(i) except Exception as e: print(f"特征预处理失败[{i}]: {e}") if not tensors: return results batch = np.stack(tensors, axis=0) try: feats = self._infer_features(batch) for j, idx in enumerate(indices): feat = feats[j] if np.isnan(feat).any() or np.isinf(feat).any(): results[idx] = None else: results[idx] = feat except Exception as e: print(f"批量特征提取失败: {e}") return results def _create_thumbnail(self, image_path, size=(60, 60)): try: thumbnail_name = f"{hash(image_path)}_{os.path.basename(image_path)}" thumbnail_path = os.path.join(THUMBNAIL_DIR, thumbnail_name) with open(image_path, 'rb') as f: img = Image.open(f).convert("RGB") img.thumbnail(size, Image.Resampling.LANCZOS) img.save(thumbnail_path, "PNG") return thumbnail_path except Exception as e: print(f"生成缩略图失败: {e}") return None def _load_thumbnail(self, thumbnail_path, size=(60, 60)): if not thumbnail_path or not os.path.exists(thumbnail_path): return None try: with open(thumbnail_path, 'rb') as f: img = Image.open(f).convert("RGB") img.thumbnail(size, Image.Resampling.LANCZOS) return ImageTk.PhotoImage(img) except Exception as e: print(f"加载缩略图失败: {e}") return None def add_template(self, part_type, image_paths): if not isinstance(image_paths, list): if isinstance(image_paths, (str, tuple)): image_paths = list(image_paths) else: image_paths = [image_paths] added = 0 add_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") with self.lock: for img_path in image_paths: feat = self._extract_image_feature(img_path) if feat is not None: thumbnail_path = self._create_thumbnail(img_path) # 避免同名文件互相覆盖：菜名 + 短哈希 stem = Path(img_path).stem suffix = Path(img_path).suffix or ".jpg" digest = abs(hash(os.path.abspath(img_path))) % 10_000_000 safe_name = f"{stem}_{digest}{suffix}" template_img_path = os.path.join(TEMPLATE_SAVE_DIR, safe_name) shutil.copy2(img_path, template_img_path) record = self.repo.insert_template( part_type, template_img_path, feat, thumbnail_path, add_time=add_time, ) self.template_library[part_type].append(record) # 学菜成功：上传到云端 template/{菜名}/ self._upload_template_cloud_async(part_type, template_img_path) added += 1 if added: self._rebuild_sim_index() if added > 0: self._notify_learned_async(part_type) return added, len(image_paths) @staticmethod def _upload_template_cloud_async(dish_name, image_path): """待学习/学菜写入本地后，异步上传到服务器 template/{菜名}/。""" try: from service.template_upload import upload_template_image_async upload_template_image_async(image_path, dish_name) except Exception as exc: print(f'[WARN] schedule template upload failed: {exc}') @staticmethod def _notify_learned_async(dish_name): name = (dish_name or '').strip() if not name: return def _work(): try: from service.mgmt_client import notify_dish_learned notify_dish_learned(name) except Exception as exc: print(f'[WARN] notify learned failed: {exc}') threading.Thread(target=_work, daemon=True).start() def add_templates_batch(self, pairs, progress_callback=None): """批量导入 [(菜名, 图片路径), ...]，只在最后同步一次，避免每张都 pull 冲掉本地库。 :return: (成功数, 失败列表[(path, reason), ...]) """ prepared = [] failures = [] total = len(pairs) for i, (part_type, img_path) in enumerate(pairs, 1): if progress_callback: progress_callback(i, total, part_type, img_path) name = (part_type or "").strip() if not name: failures.append((img_path, "空菜名")) continue if len(name) > 100: name = name[:100] if not os.path.isfile(img_path): failures.append((img_path, "文件不存在")) continue try: feat = self._extract_image_feature(img_path) except Exception as e: failures.append((img_path, f"特征异常: {e}")) continue if feat is None: failures.append((img_path, "特征提取失败")) continue try: thumbnail_path = self._create_thumbnail(img_path) stem = Path(img_path).stem suffix = Path(img_path).suffix or ".jpg" digest = abs(hash(os.path.abspath(img_path))) % 10_000_000 safe_name = f"{stem}_{digest}{suffix}" template_img_path = os.path.join(TEMPLATE_SAVE_DIR, safe_name) shutil.copy2(img_path, template_img_path) prepared.append({ "category_name": name, "image_url": template_img_path, "feature": np.asarray(feat, dtype=np.float32).tolist(), "thumbnail_url": thumbnail_path or "", }) except Exception as e: failures.append((img_path, f"保存失败: {e}")) if prepared: self.repo.insert_templates_batch(prepared) self.load_templates() # 批量导入：对成功写入的菜名去重后上报「已学习」+ 上传云端模板目录 learned_names = sorted({item['category_name'] for item in prepared if item.get('category_name')}) for name in learned_names: self._notify_learned_async(name) for item in prepared: self._upload_template_cloud_async( item.get('category_name'), item.get('image_url'), ) return len(prepared), failures def delete_template(self, part_type, template_index): if part_type in self.template_library and 0 <= template_index < len(self.template_library[part_type]): template_to_delete = self.template_library[part_type][template_index] deleted = self.repo.delete_template_by_id(template_to_delete.get('id')) if deleted: try: if deleted.get('path') and os.path.exists(deleted['path']): os.remove(deleted['path']) except Exception as e: print(f"删除模板图片失败: {e}") try: if deleted.get('thumbnail') and os.path.exists(deleted['thumbnail']): os.remove(deleted['thumbnail']) except Exception as e: print(f"删除缩略图失败: {e}") with self.lock: del self.template_library[part_type][template_index] if not self.template_library[part_type]: del self.template_library[part_type] self._rebuild_sim_index() return True return False def delete_template_category(self, part_type): if part_type in self.template_library: removed = self.repo.delete_category(part_type) for template in removed: try: if template.get('thumbnail') and os.path.exists(template['thumbnail']): os.remove(template['thumbnail']) except Exception as e: print(f"删除缩略图失败: {e}") with self.lock: if part_type in self.template_library: del self.template_library[part_type] self._rebuild_sim_index() return True return False def _rank_categories(self, feature, allowed_names=None): """按类别聚合最高相似度，返回排序后的候选列表。 allowed_names: 若给定，只在该菜名集合内匹配（小模版库）。 """ feature = np.asarray(feature, dtype=np.float32).reshape(-1) allow = None if allowed_names is not None: allow = {str(n).strip() for n in allowed_names if str(n).strip()} if not allow: return [], None, -1.0 with self.lock: matrix = self._sim_matrix categories = self._sim_categories if matrix.shape[0] == 0: return [], None, -1.0 scores = matrix @ feature # (T,) category_best = {} for cat, sim in zip(categories, scores): if allow is not None and cat not in allow: continue sim_f = float(sim) if sim_f > category_best.get(cat, -1.0): category_best[cat] = sim_f ranked = sorted(category_best.items(), key=lambda x: x[1], reverse=True) return ranked, (ranked[0][0] if ranked else None), (ranked[0][1] if ranked else -1.0) def staple_candidate_names(self): """主食小库候选：配置名单 ∩ 本地已有模板的菜名。""" try: from dao.settings import DISH_STAPLE_NAMES configured = {str(n).strip() for n in (DISH_STAPLE_NAMES or []) if str(n).strip()} except Exception: configured = set() if not configured: return set() with self.lock: available = {str(k).strip() for k in (self.template_library or {}).keys() if str(k).strip()} return configured & available @staticmethod def _score_hit(best, score, hard_thr, soft_thr): """是否达到认定阈值。返回 (hit, how_suffix)。""" if best is None: return False, "" if score >= float(hard_thr): return True, "" if score > float(soft_thr): return True, "·软阈值" return False, "" @staticmethod def _match_filter_names(): """模板比对后需丢弃的菜名（如手、空盘）。""" try: from dao.settings import DISH_MATCH_FILTER_NAMES return {str(n).strip() for n in (DISH_MATCH_FILTER_NAMES or []) if str(n).strip()} except Exception: return {"手", "空盘"} def _filtered_match_result(self, name, score, feat, top_k=None, scope="filtered"): print(f"[match] filtered template hit: {name} score={score}") return { "text": f"已过滤：{name}", "is_unknown": False, "is_filtered": True, "label": None, "score": float(score) if score is not None else None, "suggested_category": name, "filtered_as": name, "top_k": top_k or [], "pending_id": None, "feature": feat, "match_scope": scope, } def predict_from_feature(self, feat, image_path=None, auto_pending=True): """已有特征时直接匹配（避免待学习提交时重复提特征）。 仅在「当前吃饭餐段」内：主食小库 ∪ 当前餐段菜品 合并比对（主食略优先）； 不在餐段 / 无今日排餐：全部未知（不进主食、不进大库）。 均未达标 → 未知，不回退大模板库。 """ if feat is None: return { "text": "特征提取失败", "error": "特征提取失败", "is_unknown": True, "label": None, "score": None, "suggested_category": None, "top_k": [], "pending_id": None, "feature": None, } period_names = None period_only = False small_thr = 0.80 soft_thr = 0.75 staple_thr = 0.80 staple_soft = 0.75 staple_margin = 0.03 match_scope = "big" try: from dao.meal_schedule import get_meal_schedule_store from dao.settings import ( DISH_MEAL_SMALL_LIB_SOFT_THRESHOLD, DISH_MEAL_SMALL_LIB_THRESHOLD, DISH_STAPLE_LIB_SOFT_THRESHOLD, DISH_STAPLE_LIB_THRESHOLD, DISH_STAPLE_PRIORITY_MARGIN, ) store = get_meal_schedule_store() # 必须「在当前餐段时间」才识别；有今日排餐但不在餐段 → 仍全部未知 if store.should_use_small_library() and store.current_period(): period_only = True period_names = store.current_period_dish_names() small_thr = float(DISH_MEAL_SMALL_LIB_THRESHOLD) soft_thr = float(DISH_MEAL_SMALL_LIB_SOFT_THRESHOLD) staple_thr = float(DISH_STAPLE_LIB_THRESHOLD) staple_soft = float(DISH_STAPLE_LIB_SOFT_THRESHOLD) staple_margin = float(DISH_STAPLE_PRIORITY_MARGIN) except Exception as exc: print(f'[WARN] meal period-menu lookup failed: {exc}') period_names = None period_only = False # ---------- 1) 在当前餐段：主食 ∪ 当前餐段 合并候选 ---------- if period_only: staple_names = self.staple_candidate_names() period_set = set(period_names or []) filter_names = self._match_filter_names() # 过滤类若本地有模板，纳入候选：先比中再丢弃，避免误认成其他菜 with self.lock: lib_names = {str(k).strip() for k in (self.template_library or {}).keys() if str(k).strip()} filter_in_lib = filter_names & lib_names allowed = set(staple_names) | period_set | filter_in_lib ranked, _, _ = ( self._rank_categories(feat, allowed_names=allowed) if allowed else ([], None, -1.0) ) # 从合并结果里拆出主食 / 餐段 / 过滤类各自最高分 best_st, max_st = None, -1.0 best_p, max_p = None, -1.0 best_f, max_f = None, -1.0 for name, score in ranked: sc = float(score) if name in filter_in_lib and sc > max_f: best_f, max_f = name, sc if name in staple_names and sc > max_st: best_st, max_st = name, sc if name in period_set and sc > max_p: best_p, max_p = name, sc # 过滤类达标：直接丢弃该框（不再与菜品比优先） f_hit, _ = self._score_hit(best_f, max_f, staple_thr, staple_soft) # 过滤类用与主食相同阈值；也可用餐段阈值取较松者 if not f_hit: f_hit, _ = self._score_hit(best_f, max_f, small_thr, soft_thr) st_hit, st_how = self._score_hit(best_st, max_st, staple_thr, staple_soft) p_hit, p_how = self._score_hit(best_p, max_p, small_thr, soft_thr) top_k = [{"category": n, "score": s} for n, s in (ranked or [])[:3]] if f_hit and best_f is not None: # 过滤类分不低于当前菜/主食命中分 → 认定应丢弃 other_best = -1.0 if st_hit: other_best = max(other_best, float(max_st)) if p_hit: other_best = max(other_best, float(max_p)) if float(max_f) >= other_best - float(staple_margin): return self._filtered_match_result( str(best_f).strip(), float(max_f), feat, top_k=top_k, scope="filtered", ) chosen = None how = "" scope = "period_miss" # 主食达标，且不低于餐段菜太多 → 优先主食 if st_hit and (not p_hit or float(max_st) >= float(max_p) - float(staple_margin)): chosen, how, scope = best_st, f"主食+餐段{st_how}", "staple" elif p_hit: chosen, how, scope = best_p, f"主食+餐段{p_how}", "small" elif st_hit: chosen, how, scope = best_st, f"主食+餐段{st_how}", "staple" if chosen is not None: score = float(max_st) if scope == "staple" else float(max_p) if str(chosen).strip() in filter_names: return self._filtered_match_result( str(chosen).strip(), score, feat, top_k=top_k, scope=scope, ) text = f"{chosen} 相似度：{score:.3f}（{how}）" return { "text": text, "is_unknown": False, "label": chosen, "score": score, "suggested_category": chosen, "top_k": top_k, "pending_id": None, "feature": feat, "match_scope": scope, } sug, sug_score = None, -1.0 if best_f is not None and float(max_f) > sug_score: sug, sug_score = best_f, float(max_f) if best_st is not None and float(max_st) > sug_score: sug, sug_score = best_st, float(max_st) if best_p is not None and float(max_p) > sug_score: sug, sug_score = best_p, float(max_p) # 最接近手/空盘：整框丢弃，不进未知/待学习 if sug is not None and str(sug).strip() in filter_names: return self._filtered_match_result( str(sug).strip(), float(sug_score), feat, top_k=top_k, scope="period_miss", ) pending_id = None if auto_pending and image_path: pending_id = self.pending_store.add(image_path, feat, sug, sug_score) if sug is not None: text = ( f"未知菜品 (相似度{sug_score:.3f}，最接近：{sug}，" f"不在主食∪当前餐段或未达阈值，已记录待学习)" ) else: text = "未知菜品 (主食∪当前餐段均无匹配模板，已记录待学习)" return { "text": text, "is_unknown": True, "label": None, "score": sug_score if sug is not None else None, "suggested_category": sug, "top_k": top_k, "pending_id": pending_id, "feature": feat, "match_scope": "period_miss", } # ---------- 2) 不在餐段 / 无排餐：不进主食、不进大库，全部未知 ---------- pending_id = None if auto_pending and image_path: pending_id = self.pending_store.add(image_path, feat, None, None) return { "text": "未知菜品 (不在当前餐段，不匹配主食/大库，已记录待学习)", "is_unknown": True, "label": None, "score": None, "suggested_category": None, "top_k": [], "pending_id": pending_id, "feature": feat, "match_scope": "no_menu", } def predict_detail(self, image_path, auto_pending=True): feat = self._extract_image_feature(image_path) path_for_pending = image_path if isinstance(image_path, str) else None return self.predict_from_feature(feat, image_path=path_for_pending, auto_pending=auto_pending) def predict(self, image_path): return self.predict_detail(image_path)['text'] def get_pending_list(self): return self.pending_store.get_list(PendingLearningStore.STATUS_PENDING) def get_pending_count(self): return self.pending_store.count(PendingLearningStore.STATUS_PENDING) def learn_from_pending(self, pending_id, category): """确认学习一条待学习记录（只处理这一条，不轮询其余待学习项）。""" item = self.pending_store.get(pending_id) if not item: return False img_path = item.get("image_path") if not img_path or not os.path.exists(img_path): return False added, _ = self.add_template(category, [img_path]) if added <= 0: return False self.pending_store.mark_learned(pending_id, category) return True def ignore_pending(self, pending_id): """忽略一条：删本地图并从待学习库移除。""" return self.pending_store.mark_ignored(pending_id) def ignore_all_pending(self): """忽略全部 pending：删图 + 一次写盘，避免全部忽略时卡死。""" return self.pending_store.ignore_all_pending() def delete_pending(self, pending_id): return self.pending_store.delete(pending_id) def get_template_list(self, load_thumbs=False): """返回 [(菜名, 数量, 缩略图路径或PhotoImage|None), ...]。 load_thumbs=False（默认）只返回缩略图路径，避免一次创建上千个 PhotoImage 导致卡死/闪退。 """ template_list = [] with self.lock: for k, v in self.template_library.items(): if v: thumb = v[0].get('thumbnail') or '' if load_thumbs: thumb = self._load_thumbnail(thumb, size=(42, 42)) template_list.append((k, len(v), thumb)) else: template_list.append((k, 0, None)) return template_list def get_template_details(self, part_type, load_thumbs=False): details = [] with self.lock: raw_details = self.template_library.get(part_type, []) for item in raw_details: item_copy = item.copy() if load_thumbs: item_copy['thumbnail_obj'] = self._load_thumbnail(item.get('thumbnail')) else: item_copy['thumbnail_obj'] = None details.append(item_copy) return details def clear_all(self): self.repo.clear_all() with self.lock: self.template_library.clear() self._rebuild_sim_index() self.pending_store.clear_all() for file in os.listdir(THUMBNAIL_DIR): file_path = os.path.join(THUMBNAIL_DIR, file) try: if os.path.isfile(file_path): os.unlink(file_path) except Exception as e: print(f"删除缩略图失败: {e}") def set_threshold(self, threshold): with self.lock: self.threshold = threshold self.repo.set_threshold(threshold) def save_templates(self): """持久化阈值等配置（模板已实时写入数据库）。""" try: self.repo.set_threshold(self.threshold) print(f"[OK] config saved to database ({self.repo.backend_name()})") except Exception as e: print(f"[ERR] save config failed: {e}") def load_templates(self): try: threshold, library = self.repo.load_library() with self.lock: self.threshold = threshold self.template_library = defaultdict(list, library) self._rebuild_sim_index() print(f"[OK] loaded {len(self.template_library)} categories from database ({self.repo.backend_name()})") except Exception as e: print(f"[ERR] load templates failed: {e}") # -------------------------- # UI基础父类（公共UI逻辑） # -------------------------- class BaseClassifierUI(PendingLearningUIMixin): def __init__(self, root): self.root = root self.root.title("图像自动分类工具") self.root.state('zoomed') self.root.encoding = 'utf-8' setup_ttk_styles() self.manager = PartTemplateManager(0.7) self.current_img = None self.detect_img = None self.class_name = tk.StringVar(value="产品A") self.thresh_val = tk.DoubleVar(value=self.manager.threshold) # 车牌检测导入 from core.detect_plate_process import Config, YOLOv11PlateExtractor self.cfg = Config() self.app = YOLOv11PlateExtractor(self.cfg) self.create_ui() def create_ui(self): main_paned = ttk.PanedWindow(self.root, orient=tk.HORIZONTAL) main_paned.pack(fill=tk.BOTH, expand=True, padx=10, pady=10) left_frame = ttk.Frame(main_paned, width=300) main_paned.add(left_frame) ttk.Label(left_frame, text="📋 模板库（缩略图预览）", style='Title.TLabel').pack(pady=5, anchor=tk.W) self.template_frame = ttk.Frame(left_frame) self.template_frame.pack(fill=tk.BOTH, expand=True, pady=5) canvas = tk.Canvas(self.template_frame) scrollbar = ttk.Scrollbar(self.template_frame, orient="vertical", command=canvas.yview) self.scrollable_frame = ttk.Frame(canvas) self.scrollable_frame.bind("", lambda e: canvas.configure(scrollregion=canvas.bbox("all"))) canvas.create_window((0, 0), window=self.scrollable_frame, anchor="nw") canvas.configure(yscrollcommand=scrollbar.set) canvas.pack(side="left", fill=tk.BOTH, expand=True) scrollbar.pack(side="right", fill=tk.Y) ttk.Button(left_frame, text="🗑️ 清空模板库", command=self.clear_lib, style='Large.TButton').pack(fill=tk.X, pady=3) self.setup_pending_learning_ui(left_frame) right_frame = ttk.Frame(main_paned) main_paned.add(right_frame) top_frame = ttk.Frame(right_frame) top_frame.pack(fill=tk.X, pady=5) ttk.Label(top_frame, text="相似度阈值：", style='Medium.TLabel').pack(side=tk.LEFT, padx=5) thresh_spin = ttk.Spinbox(top_frame, from_=0.0, to=1.0, increment=0.01, textvariable=self.thresh_val, width=10, style='Large.TSpinbox') thresh_spin.pack(side=tk.LEFT) ttk.Button(top_frame, text="应用", style='Large.TButton', command=lambda: self.manager.set_threshold(self.thresh_val.get())).pack(side=tk.LEFT, padx=5) add_frame = ttk.Frame(right_frame) add_frame.pack(fill=tk.X, pady=5) ttk.Label(add_frame, text="类别名称：", style='Medium.TLabel').pack(side=tk.LEFT, padx=5) cls_entry = ttk.Entry(add_frame, textvariable=self.class_name, width=15, style='Large.TEntry') cls_entry.pack(side=tk.LEFT, padx=5) ttk.Button(add_frame, text="📁 添加模板图片", style='Large.TButton', command=self.add_template).pack(side=tk.LEFT, padx=5) # 相机按钮，子类实例化时再修改状态 self.camera_add_btn = ttk.Button(add_frame, text="📷 拍照添加模板", style='Large.TButton') self.camera_add_btn.pack(side=tk.LEFT, padx=5) func_frame = ttk.LabelFrame(right_frame, text="核心功能") func_frame.pack(fill=tk.X, pady=8) ttk.Button(func_frame, text="📂 选择本地图片识别", style='Large.TButton', command=self.choose_and_predict_img).pack(side=tk.LEFT, padx=5, pady=5) self.camera_btn = ttk.Button(func_frame, text="📷 相机拍照识别", style='Large.TButton') self.camera_btn.pack(side=tk.LEFT, padx=5, pady=5) self.camera_status_label = ttk.Label(func_frame, text="相机状态：未连接", foreground="red", style='Medium.TLabel') self.camera_status_label.pack(side=tk.LEFT, padx=10, pady=5) img_frame = ttk.LabelFrame(right_frame, text="图片预览") img_frame.pack(fill=tk.BOTH, expand=True, pady=5) self.img_label = ttk.Label(img_frame, text="暂无图片", style='Medium.TLabel') self.img_label.pack(pady=50) res_frame = ttk.LabelFrame(right_frame, text="识别结果") res_frame.pack(fill=tk.X, pady=5) self.result_label = ttk.Label(res_frame, text="程序启动中...", style='Title.TLabel', foreground="blue") self.result_label.pack(pady=8) self.refresh_template_list() def refresh_template_list(self): for widget in self.scrollable_frame.winfo_children(): widget.destroy() items = self.manager.get_template_list(load_thumbs=False) if not items: ttk.Label(self.scrollable_frame, text="暂无模板，请先添加", style='Medium.TLabel').pack(pady=20) return for idx, (name, count, _thumbnail) in enumerate(items, 1): if count == 0: continue item_frame = ttk.Frame(self.scrollable_frame, relief=tk.GROOVE, borderwidth=1) item_frame.pack(fill=tk.X, pady=3, padx=2) text_frame = ttk.Frame(item_frame) text_frame.pack(side=tk.LEFT, fill=tk.X, expand=True, padx=5) ttk.Label(text_frame, text=f"{idx}. {name}", style='Title.TLabel').pack(anchor=tk.W) ttk.Label(text_frame, text=f"模板数量：{count}张", style='Small.TLabel', foreground="gray").pack(anchor=tk.W) delete_btn = ttk.Button(item_frame, text="删除", style='Large.TButton', command=lambda n=name: self.delete_template_category(n)) delete_btn.pack(side=tk.RIGHT, padx=5, pady=5) item_frame.bind("", lambda e, n=name: self.show_template_detail(n)) for child in text_frame.winfo_children(): child.bind("", lambda e, n=name: self.show_template_detail(n)) def show_template_detail(self, part_type): template_details = self.manager.get_template_details(part_type, load_thumbs=False) if not template_details: messagebox.showinfo("提示", "该类别暂无模板信息") return detail_win = tk.Toplevel(self.root) detail_win.title(f"模板详情 - {part_type}") detail_win.geometry("1400x900") self.center_window(detail_win, 1400, 900) detail_win.transient(self.root) detail_win.grab_set() detail_win._detail_photos = [] main_frame = ttk.Frame(detail_win, padding=10) main_frame.pack(fill=tk.BOTH, expand=True) ttk.Label(main_frame, text=f"类别：{part_type} (共{len(template_details)}张模板)", style='Title.TLabel').pack(pady=5) canvas = tk.Canvas(main_frame) scrollbar = ttk.Scrollbar(main_frame, orient="vertical", command=canvas.yview) scrollable_frame = ttk.Frame(canvas) scrollable_frame.bind("", lambda e: canvas.configure(scrollregion=canvas.bbox("all"))) canvas.create_window((0, 0), window=scrollable_frame, anchor="nw") canvas.configure(yscrollcommand=scrollbar.set) canvas.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) scrollbar.pack(side=tk.RIGHT, fill=tk.Y) for idx, temp in enumerate(template_details, 1): frame = ttk.LabelFrame(scrollable_frame, text=f"模板 {idx}", padding=10) frame.pack(fill=tk.X, pady=8) info_delete_frame = ttk.Frame(frame) info_delete_frame.pack(fill=tk.X, pady=2) info_frame = ttk.Frame(info_delete_frame) info_frame.pack(side=tk.LEFT, fill=tk.X, expand=True) ttk.Label(info_frame, text=f"添加时间：{temp.get('add_time')}", style='Medium.TLabel').pack(anchor=tk.W) ttk.Label(info_frame, text=f"文件路径：{temp.get('path')}", style='Small.TLabel').pack(anchor=tk.W) delete_btn = ttk.Button(info_delete_frame, text="删除此模板", style='Large.TButton', command=lambda p=part_type, i=idx - 1: self.delete_single_template(p, i)) delete_btn.pack(side=tk.RIGHT, padx=5) try: with open(temp['path'], 'rb') as f: img = Image.open(f).convert("RGB") img.thumbnail((800, 500) if idx == 1 else (240, 180), Image.Resampling.LANCZOS) photo = ImageTk.PhotoImage(img) detail_win._detail_photos.append(photo) img_label = ttk.Label(frame, image=photo) img_label.image = photo img_label.pack(pady=5) except Exception as e: ttk.Label(frame, text=f"图片加载失败：{str(e)}", foreground="red", style='Small.TLabel').pack(pady=2) def center_window(self, window, width, height): screen_width = window.winfo_screenwidth() screen_height = window.winfo_screenheight() x = (screen_width - width) // 2 y = (screen_height - height) // 2 window.geometry(f"{width}x{height}+{x}+{y}") 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, is_category=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, is_category=False)) threading.Thread(target=delete_async, daemon=True).start() def _handle_delete_result(self, success, part_type, is_category): if success: if is_category: messagebox.showinfo("成功", f"已删除类别「{part_type}」的所有模板") else: messagebox.showinfo("成功", "模板删除成功") 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 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: messagebox.showinfo("成功", f"添加完成！\n成功：{added}张 | 总数：{total}张")) 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.result_label.config(text="模板库已清空", foreground="black") def dish_box_label(self, predict_text): """从识别文案提取框上展示的菜品名（认菜不认盘）。""" if not predict_text or predict_text == "特征提取失败": return "识别失败" if predict_text.startswith("未知"): score = predict_text.split("：")[-1] if "：" in predict_text else "" return f"未知 {score}".strip() if "相似度" in predict_text: name = predict_text.split(" 相似度")[0].strip() score = predict_text.split("：")[-1] if "：" in predict_text else "" return f"{name} {score}".strip() return predict_text def show_preview_image(self, image_path): with open(image_path, 'rb') as f: im = Image.open(f).convert("RGB") im.thumbnail((1800, 800)) photo = ImageTk.PhotoImage(im) self.img_label.config(text="", image=photo) self.img_label.image = photo def build_dish_labeled_image(self, source_image_path, boxes, dish_labels, styles=None): from pathlib import Path if isinstance(source_image_path, np.ndarray): stem = f"frame_{int(datetime.datetime.now().timestamp() * 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 choose_and_predict_img(self): file_path = filedialog.askopenfilename(title="选择识别图片", filetypes=[("图片", "*.jpg *.jpeg *.png *.bmp"), ("所有文件", "*.*")]) if not file_path: return visualize_save_path, cropped_file_paths, boxes = self.app.process_single_image(file_path) if not cropped_file_paths: messagebox.showwarning("提示", "未检测到餐盘区域") return final_result, dish_labels, unknown_count, labeled_path, dishes = self.run_plate_predict_batch( file_path, cropped_file_paths, boxes, progress_callback=lambda cur, _total: self.result_label.config( text=f"正在识别第 {cur} 张...", foreground="orange" ), ) if labeled_path: self.show_preview_image(labeled_path) elif visualize_save_path: self.show_preview_image(visualize_save_path) self.result_label.config(text=final_result or "无结果", foreground="green") self.confirm_unknown_dishes(dishes) # 以下方法由子类USB/海康分别实现 def auto_connect_camera(self): raise NotImplementedError("子类实现") def take_photo(self): raise NotImplementedError("子类实现") def camera_capture_and_add_template(self): raise NotImplementedError("子类实现") def camera_capture_and_predict(self): raise NotImplementedError("子类实现") def close_camera(self): raise NotImplementedError("子类实现")