- src/: 最终交付物(solve.py base64 API、method_l_shape.py 核心算法、verify_result.py 验证工具) - docs/: 方案文档与实验演进记录 - try/: 历史实验脚本(A~K 方法) - 10/10 样本求解成功,3 个独立真值锚点偏差 <=4px
194 lines
7.2 KiB
Python
194 lines
7.2 KiB
Python
"""方法A2:蒙版 CCOEFF_NORMED 多尺度匹配 + 跨尺度峰显著度。
|
|
|
|
修正方法A的两个问题:
|
|
1. CCORR_NORMED 不减均值 → 换 CCOEFF_NORMED(有蒙版),区分度更高
|
|
2. 跨尺度直接比分数不公平(小模板占便宜)→ 每尺度算峰显著度 (max-median)/std
|
|
|
|
输出:
|
|
- 每对:1.0 尺度参考、最优显著度尺度、该尺度 top-3 峰(NMS)→ 验证干扰结构
|
|
- 可视化:mark | top-3 峰局部裁剪 并排图 → try/out/A2/
|
|
- 汇总表 → try/out/A2/summary.txt
|
|
"""
|
|
from pathlib import Path
|
|
import cv2
|
|
import numpy as np
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
CAP = ROOT / "captchas"
|
|
OUT = ROOT / "try" / "out" / "A2"
|
|
OUT.mkdir(parents=True, exist_ok=True)
|
|
|
|
SCALES = np.arange(0.50, 1.51, 0.05)
|
|
TOP_K = 3
|
|
|
|
|
|
def load_mark(path):
|
|
"""返回 (BGR float32, alpha float32 0~1) 或 (None, None)。"""
|
|
try:
|
|
m = cv2.imread(str(path), cv2.IMREAD_UNCHANGED)
|
|
if m is None or m.ndim < 2:
|
|
return None, None
|
|
if m.ndim == 3 and m.shape[2] == 4:
|
|
return m[..., :3].astype(np.float32), m[..., 3].astype(np.float32) / 255.0
|
|
return m.astype(np.float32), np.ones(m.shape[:2], np.float32)
|
|
except Exception as e:
|
|
print(f" mark 读取失败: {e}")
|
|
return None, None
|
|
|
|
|
|
def match_ccoeff(big, tpl_f, mask_f):
|
|
"""蒙版 CCOEFF_NORMED。返回响应图或 None。"""
|
|
try:
|
|
res = cv2.matchTemplate(big, tpl_f, cv2.TM_CCOEFF_NORMED, mask=mask_f)
|
|
return np.nan_to_num(res, nan=-1.0, posinf=-1.0, neginf=-1.0).astype(np.float32)
|
|
except cv2.error as e:
|
|
print(f" 匹配失败: {e}")
|
|
return None
|
|
|
|
|
|
def scale_response(big, bgr_f, alpha_f, scale):
|
|
"""单尺度匹配,返回 (响应图, 模板宽, 模板高);失败返回 (None, 0, 0)。"""
|
|
try:
|
|
if abs(scale - 1.0) < 1e-9:
|
|
tpl, mask = bgr_f, alpha_f
|
|
else:
|
|
w = max(12, round(bgr_f.shape[1] * scale))
|
|
h = max(12, round(bgr_f.shape[0] * scale))
|
|
if w >= big.shape[1] or h >= big.shape[0]:
|
|
return None, 0, 0
|
|
tpl = cv2.resize(bgr_f, (w, h), interpolation=cv2.INTER_AREA).astype(np.float32)
|
|
mask = cv2.resize(alpha_f, (w, h), interpolation=cv2.INTER_AREA).astype(np.float32)
|
|
res = match_ccoeff(big, tpl, mask)
|
|
if res is None:
|
|
return None, 0, 0
|
|
return res, int(mask.shape[1]), int(mask.shape[0])
|
|
except cv2.error as e:
|
|
print(f" 尺度失败(scale={scale}): {e}")
|
|
return None, 0, 0
|
|
|
|
|
|
def prominence(res):
|
|
"""峰显著度:峰高相对响应图整体的突出程度,跨尺度可比。"""
|
|
try:
|
|
mx = float(res.max())
|
|
med = float(np.median(res))
|
|
sd = float(res.std())
|
|
return (mx - med) / sd if sd > 1e-6 else 0.0
|
|
except Exception:
|
|
return 0.0
|
|
|
|
|
|
def topk_peaks(res, tpl_w, tpl_h, k=TOP_K):
|
|
"""NMS 取前 k 个峰:返回 [(score, (x, y)), ...]。"""
|
|
try:
|
|
r = res.copy()
|
|
peaks = []
|
|
for _ in range(k):
|
|
_, score, _, loc = cv2.minMaxLoc(r)
|
|
if score <= -1.0:
|
|
break
|
|
peaks.append((float(score), loc))
|
|
x0 = max(0, loc[0] - int(tpl_w * 0.7))
|
|
y0 = max(0, loc[1] - int(tpl_h * 0.7))
|
|
x1 = min(r.shape[1], loc[0] + int(tpl_w * 0.7))
|
|
y1 = min(r.shape[0], loc[1] + int(tpl_h * 0.7))
|
|
r[y0:y1, x0:x1] = -1.0
|
|
return peaks
|
|
except Exception as e:
|
|
print(f" 峰提取失败: {e}")
|
|
return []
|
|
|
|
|
|
def side_by_side(name, mark_bgr, big, peaks, tpl_w, tpl_h):
|
|
"""mark | top-k 峰裁剪 并排图。"""
|
|
try:
|
|
mark_u8 = np.clip(mark_bgr, 0, 255).astype(np.uint8)
|
|
mark_u8 = cv2.copyMakeBorder(mark_u8, 2, 2, 2, 2, cv2.BORDER_CONSTANT,
|
|
value=(255, 255, 255))
|
|
tiles = [cv2.resize(mark_u8, (tpl_w, tpl_h))]
|
|
for score, (x, y) in peaks:
|
|
x0, y0 = max(0, x), max(0, y)
|
|
x1, y1 = min(big.shape[1], x + tpl_w), min(big.shape[0], y + tpl_h)
|
|
crop = big[y0:y1, x0:x1].copy()
|
|
if crop.shape[0] != tpl_h or crop.shape[1] != tpl_w:
|
|
crop = cv2.copyMakeBorder(crop, 0, tpl_h - crop.shape[0],
|
|
0, tpl_w - crop.shape[1],
|
|
cv2.BORDER_CONSTANT, value=(128, 128, 128))
|
|
cv2.putText(crop, f"{score:.3f}", (2, 12), cv2.FONT_HERSHEY_SIMPLEX,
|
|
0.4, (0, 0, 255), 1)
|
|
tiles.append(crop)
|
|
sep = np.full((tpl_h, 4, 3), 255, np.uint8)
|
|
row = tiles[0]
|
|
for t in tiles[1:]:
|
|
row = np.hstack([row, sep, t])
|
|
cv2.imwrite(str(OUT / f"{name[:12]}-crops.png"), row)
|
|
except Exception as e:
|
|
print(f" 可视化失败: {e}")
|
|
|
|
|
|
def process_pair(name, lines):
|
|
try:
|
|
big = cv2.imread(str(CAP / f"{name}.jpeg"), cv2.IMREAD_COLOR)
|
|
if big is None:
|
|
lines.append(f"{name} SKIP 大图读取失败")
|
|
return
|
|
big_f = big.astype(np.float32)
|
|
bgr_f, alpha_f = load_mark(CAP / f"{name}-mark.png")
|
|
if bgr_f is None:
|
|
lines.append(f"{name} SKIP mark 读取失败")
|
|
return
|
|
|
|
# 参考:1.0 尺度
|
|
res10, _, _ = scale_response(big_f, bgr_f, alpha_f, 1.0)
|
|
if res10 is not None:
|
|
_, _, _, loc10 = cv2.minMaxLoc(res10)
|
|
ref = f"1.0尺度峰={float(res10.max()):.3f}@{loc10}"
|
|
else:
|
|
ref = "1.0尺度N/A"
|
|
|
|
# 扫尺度,选显著度最高者
|
|
best = None # (显著度, 尺度, 响应图, 模板宽, 模板高)
|
|
curve = []
|
|
for sc in SCALES:
|
|
res, tw_, th_ = scale_response(big_f, bgr_f, alpha_f, float(sc))
|
|
if res is None:
|
|
curve.append(0.0)
|
|
continue
|
|
p = prominence(res)
|
|
curve.append(p)
|
|
if best is None or p > best[0]:
|
|
best = (p, float(sc), res, tw_, th_)
|
|
curve = np.array(curve)
|
|
if best is None:
|
|
lines.append(f"{name} SKIP 无有效尺度")
|
|
return
|
|
p, sc, res, tw, th = best
|
|
peaks = topk_peaks(res, tw, th)
|
|
|
|
top = " ".join(f"#{i+1}:{s:.3f}@{loc}" for i, (s, loc) in enumerate(peaks))
|
|
print(f"{name[:12]} {ref} 最优尺度={sc:.2f} 显著度={p:.2f} {top}")
|
|
lines.append(f"{name[:12]}\t{ref}\t尺度={sc:.2f}\t显著度={p:.2f}\t{top}")
|
|
side_by_side(name, bgr_f, big, peaks, tw, th)
|
|
# 大图上画 top-3 框
|
|
vis = big.copy()
|
|
colors = [(0, 0, 255), (0, 165, 255), (255, 0, 0)]
|
|
for i, (s, (x, y)) in enumerate(peaks):
|
|
cv2.rectangle(vis, (x, y), (x + tw, y + th), colors[i % 3], 2)
|
|
cv2.imwrite(str(OUT / f"{name[:12]}-match.png"), vis)
|
|
except Exception as e:
|
|
print(f"{name[:12]} 处理失败: {e}")
|
|
lines.append(f"{name[:12]} ERROR {e}")
|
|
|
|
|
|
def main():
|
|
lines = []
|
|
pairs = sorted({p.name.replace("-mark.png", "") for p in CAP.glob("*-mark.png")})
|
|
for name in pairs:
|
|
process_pair(name, lines)
|
|
(OUT / "summary.txt").write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
print(f"\n结果与 summary 已写入 {OUT}/")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|