PokemonCardSearch/code/tools/audit_detect_all.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
真实拍摄图多卡检测审计（249 上跑）：对比"关闭合并 vs 开启合并"的检出差异，
校准 CARD_MERGE_DIST_PX 阈值。

数据源二选一：
  --minio N    MinIO grading/capp_img_data/ 最新 N 张（C 端真实拍摄图，默认 50）
  --dir PATH   本地目录（如 data/query_images/）

每张图输出：关闭合并的原始检出数 → 开启合并后检出数、合并发生明细
（谁吸收了谁、中心距、conf）。结尾汇总：合并触发的图占比、误杀风险（原始多实例
但合并后变少的例子需要人工过目）。

用法（pytorch env）:
  python tools/audit_detect_all.py --minio 50
  python tools/audit_detect_all.py --dir data/query_images
"""
import os
import sys
import argparse

sys.stdout.reconfigure(encoding="utf-8")
_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, _ROOT)

import numpy as np
from PIL import Image

import config
from modules.yolo_detector import CardDetector


def _fetch_minio(limit, out_dir, prefix=None):
    """拉 grading 桶指定前缀最新 limit 张到 out_dir，返回本地图路径列表。"""
    import config as _cfg
    cfg = _cfg.CAPP_MINIO
    from minio import Minio
    client = Minio(cfg["endpoint"], access_key=cfg["access_key"],
                   secret_key=cfg["secret_key"], secure=bool(cfg.get("secure")))
    objs = []
    for o in client.list_objects(cfg["bucket"], prefix=(prefix or cfg["prefix"]) + "/", recursive=True):
        if o.object_name.lower().endswith((".jpg", ".jpeg", ".png", ".webp")):
            objs.append(o)
    objs.sort(key=lambda o: o.last_modified or 0, reverse=True)
    os.makedirs(out_dir, exist_ok=True)
    paths = []
    for o in objs[:limit]:
        dst = os.path.join(out_dir, os.path.basename(o.object_name))
        if not os.path.exists(dst) or os.path.getsize(dst) == 0:
            try:
                client.fget_object(cfg["bucket"], o.object_name, dst)
            except Exception as e:
                print(f"  [skip] {o.object_name}: {e}", flush=True)
                continue
        paths.append(dst)
    return paths


def _audit_one(detector, path, thr):
    """对一张图跑两遍（关/开合并），返回 dict。"""
    try:
        arr = np.array(Image.open(path).convert("RGB"))
    except Exception as e:
        return {"path": path, "error": str(e)[:100]}
    detector.merge_dist_px = 0.0
    raw = detector.detect_and_crop_all(arr)
    detector.merge_dist_px = thr
    merged = detector.detect_and_crop_all(arr)
    return {
        "path": path,
        "w": arr.shape[1], "h": arr.shape[0],
        "raw_n": len(raw),
        "merged_n": len(merged),
        "raw_confs": [round(it["conf"], 3) for it in raw],
        "merged_confs": [round(it["conf"], 3) for it in merged],
        "merged_centers": [(round(c[0]), round(c[1])) for c in (it["center"] for it in merged)],
        "merged": len(merged) < len(raw),
    }


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--minio", type=int, default=0, help="拉取 grading 桶最新 N 张")
    ap.add_argument("--prefix", type=str, default=None,
                    help="MinIO 对象前缀（默认 config.CAPP_MINIO.prefix=capp_img_data；"
                         "拍摄图更多可用 raspi_img_data）")
    ap.add_argument("--dir", type=str, default=None, help="本地图片目录")
    ap.add_argument("--limit", type=int, default=50)
    args = ap.parse_args()

    if args.dir:
        paths = sorted(
            os.path.join(args.dir, f) for f in os.listdir(args.dir)
            if f.lower().endswith((".jpg", ".jpeg", ".png", ".webp", ".bmp")))[:args.limit]
    elif args.minio:
        paths = _fetch_minio(args.minio, os.path.join(config.DATA_DIR, "audit_imgs"), args.prefix)
    else:
        ap.error("需 --minio N 或 --dir PATH")

    detector = CardDetector(config.YOLO_MODEL_PATH)
    thr = float(getattr(detector, "merge_dist_px", 50.0))
    print(f"[Audit] 阈值 CARD_MERGE_DIST_PX={thr}  图片数={len(paths)}\n", flush=True)

    n_merged = n_multi = n_zero = n_err = 0
    for i, p in enumerate(paths, 1):
        r = _audit_one(detector, p, thr)
        if "error" in r:
            n_err += 1
            print(f"[{i}/{len(paths)}] {os.path.basename(p)} 读取失败: {r['error']}", flush=True)
            continue
        n_merged += int(r["merged"])
        n_multi += int(r["merged_n"] > 1)
        n_zero += int(r["merged_n"] == 0)
        flag = " ← 合并触发" if r["merged"] else ""
        print(f"[{i}/{len(paths)}] {os.path.basename(p)}  {r['w']}×{r['h']}  "
              f"raw={r['raw_n']}({r['raw_confs']}) → merged={r['merged_n']}"
              f"({r['merged_confs']})  centers={r['merged_centers']}{flag}", flush=True)

    n_ok = len(paths) - n_err
    print(f"\n[Summary] 有效 {n_ok} 张：合并触发 {n_merged}（{n_merged / max(1, n_ok):.1%}）  "
          f"多卡图 {n_multi}  未检出 {n_zero}  读取失败 {n_err}", flush=True)
    print("[Hint] 合并触发的样例请人工过目：若 raw 里本就是两张紧贴的不同卡被并成一张，"
          "属误杀 → 调小 CARD_MERGE_DIST_PX 或启用 CARD_MERGE_MIN_IOU。", flush=True)


if __name__ == "__main__":
    main()