杨豪
d15b3c5b43
- src/: 最终交付物（solve.py base64 API、method_l_shape.py 核心算法、verify_result.py 验证工具） - docs/: 方案文档与实验演进记录 - try/: 历史实验脚本（A~K 方法） - 10/10 样本求解成功，3 个独立真值锚点偏差 <=4px
248 lines
10 KiB
Python
248 lines
10 KiB
Python
"""方法C：两阶段旋转感知模板匹配（纯 OpenCV，离线）。
|
||
|
||
阶段1（定位候选）：rot=0 的尺度扫描 0.40~1.20，蒙版 CCOEFF_NORMED，
|
||
每尺度取显著度 top-3 峰，跨尺度 NMS 去重，按显著度保留 top-15。
|
||
|
||
阶段2（旋转消歧）：对每个候选位置，把 mark 旋转 -180°~180°（步长 15°）
|
||
后与该位置 patch 做 NCC → 得到旋转响应曲线。真目标的最优旋转角应≈0°
|
||
（AGENTS.md：目标无缩放无旋转差异；干扰项带旋转）。
|
||
在 |rot*| ≤ ZERO_TOL 的候选中取 NCC 最高者为目标。
|
||
|
||
输出：try/out/C/{*-match.png,*-rot.png,summary.txt}
|
||
"""
|
||
from pathlib import Path
|
||
import math
|
||
import cv2
|
||
import numpy as np
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent
|
||
CAP = ROOT / "captchas"
|
||
OUT = ROOT / "try" / "out" / "C"
|
||
OUT.mkdir(parents=True, exist_ok=True)
|
||
|
||
SCALES = np.arange(0.40, 1.21, 0.05)
|
||
ANGLES = np.arange(-180.0, 180.0, 15.0)
|
||
PEAKS_PER_SCALE = 3
|
||
CAND_KEEP = 15
|
||
ZERO_TOL = 10.0
|
||
|
||
|
||
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 template_at(bgr, alpha, scale, angle):
|
||
"""缩放(+旋转)模板与蒙版。返回 (tpl, mask, w, h) 或 (None, None, 0, 0)。"""
|
||
try:
|
||
w = max(12, round(bgr.shape[1] * scale))
|
||
h = max(12, round(bgr.shape[0] * scale))
|
||
if w < 12 or h < 12:
|
||
return None, None, 0, 0
|
||
tpl = cv2.resize(bgr, (w, h), interpolation=cv2.INTER_AREA).astype(np.float32)
|
||
mask = cv2.resize(alpha, (w, h), interpolation=cv2.INTER_AREA).astype(np.float32)
|
||
if abs(angle) > 1e-6:
|
||
M = cv2.getRotationMatrix2D((w / 2.0, h / 2.0), angle, 1.0)
|
||
cos, sin = abs(float(M[0, 0])), abs(float(M[0, 1]))
|
||
nw, nh = int(h * sin + w * cos) + 1, int(h * cos + w * sin) + 1
|
||
M[0, 2] += nw / 2.0 - w / 2.0
|
||
M[1, 2] += nh / 2.0 - h / 2.0
|
||
tpl = cv2.warpAffine(tpl, M, (nw, nh), flags=cv2.INTER_LINEAR,
|
||
borderValue=(0, 0, 0)).astype(np.float32)
|
||
mask = cv2.warpAffine(mask, M, (nw, nh), flags=cv2.INTER_LINEAR,
|
||
borderValue=0).astype(np.float32)
|
||
return tpl, mask, int(tpl.shape[1]), int(tpl.shape[0])
|
||
except cv2.error as e:
|
||
print(f" 模板变换失败: {e}")
|
||
return None, None, 0, 0
|
||
|
||
|
||
def full_match(big_f, tpl, mask):
|
||
"""整图蒙版 CCOEFF_NORMED。返回响应图或 None。"""
|
||
try:
|
||
res = cv2.matchTemplate(big_f, tpl, cv2.TM_CCOEFF_NORMED, mask=mask)
|
||
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 topk_peaks(res, tw, th, k):
|
||
"""NMS 取前 k 峰，返回 [(score, x, y)]。"""
|
||
peaks = []
|
||
try:
|
||
r = res.copy()
|
||
for _ in range(k):
|
||
_, score, _, loc = cv2.minMaxLoc(r)
|
||
if score <= -1.0:
|
||
break
|
||
peaks.append((float(score), int(loc[0]), int(loc[1])))
|
||
x0, y0 = max(0, loc[0] - int(tw * 0.7)), max(0, loc[1] - int(th * 0.7))
|
||
x1, y1 = min(r.shape[1], loc[0] + int(tw * 0.7)), min(r.shape[0], loc[1] + int(th * 0.7))
|
||
r[y0:y1, x0:x1] = -1.0
|
||
except Exception as e:
|
||
print(f" 峰提取失败: {e}")
|
||
return peaks
|
||
|
||
|
||
def patch_ncc(big_f, cx, cy, tpl, mask):
|
||
"""在以 (cx,cy) 为中心、与旋转后模板同尺寸的 patch 上算 NCC。失败返回 -1。"""
|
||
try:
|
||
h, w = tpl.shape[:2]
|
||
x0 = int(round(cx - w / 2.0))
|
||
y0 = int(round(cy - h / 2.0))
|
||
if x0 < 0 or y0 < 0 or x0 + w > big_f.shape[1] or y0 + h > big_f.shape[0]:
|
||
return -1.0
|
||
patch = big_f[y0:y0 + h, x0:x0 + w]
|
||
res = cv2.matchTemplate(patch, tpl, cv2.TM_CCOEFF_NORMED, mask=mask)
|
||
v = float(np.nan_to_num(res[0, 0], nan=-1.0))
|
||
return v
|
||
except cv2.error:
|
||
return -1.0
|
||
|
||
|
||
def rotation_curve(big_f, cx, cy, bgr, alpha, scale, angles):
|
||
"""候选位置的旋转响应曲线。返回 (best_angle, best_ncc, ncc_at_0, samples)。"""
|
||
samples = []
|
||
try:
|
||
for ang in angles:
|
||
tpl, mask, _, _ = template_at(bgr, alpha, scale, float(ang))
|
||
if tpl is None:
|
||
samples.append(-1.0)
|
||
continue
|
||
samples.append(patch_ncc(big_f, cx, cy, tpl, mask))
|
||
arr = np.array(samples)
|
||
k = int(arr.argmax())
|
||
i0 = int(np.where(np.isclose(angles, 0.0))[0][0]) if any(np.isclose(angles, 0.0)) else -1
|
||
ncc0 = float(arr[i0]) if i0 >= 0 else -1.0
|
||
return float(angles[k]), float(arr[k]), ncc0, arr
|
||
except Exception as e:
|
||
print(f" 旋转曲线失败: {e}")
|
||
return 0.0, -1.0, -1.0, np.array(samples)
|
||
|
||
|
||
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[:12]}\tSKIP\t大图读取失败")
|
||
return
|
||
big_f = big.astype(np.float32)
|
||
bgr, alpha = load_mark(CAP / f"{name}-mark.png")
|
||
if bgr is None or alpha is None:
|
||
lines.append(f"{name[:12]}\tSKIP\tmark 读取失败")
|
||
return
|
||
|
||
# 阶段1：rot=0 尺度扫描
|
||
cands = [] # (prom, scale, w, h, x, y, score0)
|
||
for sc in SCALES:
|
||
tpl, mask, tw, th = template_at(bgr, alpha, float(sc), 0.0)
|
||
if tpl is None or tw >= big.shape[1] or th >= big.shape[0]:
|
||
continue
|
||
res = full_match(big_f, tpl, mask)
|
||
if res is None:
|
||
continue
|
||
mx, med, sd = float(res.max()), float(np.median(res)), float(res.std())
|
||
prom = (mx - med) / sd if sd > 1e-6 else 0.0
|
||
for score, x, y in topk_peaks(res, tw, th, PEAKS_PER_SCALE):
|
||
cands.append((prom, float(sc), tw, th, x, y, score))
|
||
if not cands:
|
||
lines.append(f"{name[:12]}\tSKIP\t阶段1无候选")
|
||
return
|
||
|
||
# 跨尺度 NMS：中心距离 < 0.6*模板尺寸 视为同点
|
||
cands.sort(key=lambda c: (-c[0], -c[6]))
|
||
kept = []
|
||
for c in cands:
|
||
cx, cy = c[4] + c[2] / 2.0, c[5] + c[3] / 2.0
|
||
dup = False
|
||
for k in kept:
|
||
kcx, kcy = k[4] + k[2] / 2.0, k[5] + k[3] / 2.0
|
||
if math.hypot(cx - kcx, cy - kcy) < 0.6 * max(c[2], k[2]):
|
||
dup = True
|
||
break
|
||
if not dup:
|
||
kept.append(c)
|
||
if len(kept) >= CAND_KEEP:
|
||
break
|
||
|
||
# 阶段2：旋转消歧
|
||
results = [] # (rot*, ncc*, ncc0, prom, scale, w, h, x, y)
|
||
for prom, sc, tw, th, x, y, score0 in kept:
|
||
cx, cy = x + tw / 2.0, y + th / 2.0
|
||
rot_b, ncc_b, ncc0, _ = rotation_curve(big_f, cx, cy, bgr, alpha, sc, ANGLES)
|
||
results.append((rot_b, ncc_b, ncc0, prom, sc, tw, th, x, y))
|
||
|
||
ok = [r for r in results if abs(r[0]) <= ZERO_TOL and r[1] > 0.0]
|
||
ok.sort(key=lambda r: (-r[1], abs(r[0])))
|
||
results.sort(key=lambda r: -r[1])
|
||
if ok:
|
||
t = ok[0]
|
||
rot_b, ncc_b, ncc0, prom, sc, tw, th, x, y = t
|
||
print(f"{name[:12]} 目标: scale={sc:.2f} rot*={rot_b:+.0f}° "
|
||
f"ncc*={ncc_b:.3f} ncc(0°)={ncc0:.3f} offset=({x},{y}) "
|
||
f"prom={prom:.2f} | 合规候选={len(ok)}")
|
||
lines.append(f"{name[:12]}\t目标\tscale={sc:.2f}\trot*={rot_b:+.0f}\t"
|
||
f"ncc*={ncc_b:.3f}\tncc0={ncc0:.3f}\toffset=({x},{y})\t"
|
||
f"prom={prom:.2f}\t合规={len(ok)}")
|
||
else:
|
||
r = results[0]
|
||
print(f"{name[:12]} 无零旋转候选，最优旋转响应: scale={r[4]:.2f} "
|
||
f"rot*={r[0]:+.0f}° ncc*={r[1]:.3f}")
|
||
lines.append(f"{name[:12]}\t无零旋转候选\tbest_rot*={r[0]:+.0f}\t"
|
||
f"ncc*={r[1]:.3f}\tscale={r[4]:.2f}")
|
||
|
||
# 可视化：目标框 + 旋转曲线图
|
||
vis = big.copy()
|
||
for i, r in enumerate(results[:3]):
|
||
rot_b, ncc_b, _, _, sc, tw, th, x, y = r
|
||
color = (0, 0, 255) if (ok and r is ok[0]) else (255, 0, 0)
|
||
cv2.rectangle(vis, (x, y), (x + tw, y + th), color, 2)
|
||
cv2.putText(vis, f"s={sc:.2f} r={rot_b:.0f} {ncc_b:.2f}",
|
||
(x, max(12, y - 4)), cv2.FONT_HERSHEY_SIMPLEX, 0.4, color, 1)
|
||
cv2.imwrite(str(OUT / f"{name[:12]}-match.png"), vis)
|
||
if ok:
|
||
_, _, _, _, sc, _, _, _, _ = ok[0]
|
||
cx, cy = ok[0][7] + ok[0][5] / 2.0, ok[0][8] + ok[0][6] / 2.0
|
||
_, _, _, curve = rotation_curve(big_f, cx, cy, bgr, alpha, sc, ANGLES)
|
||
img = np.full((160, 480, 3), 255, np.uint8)
|
||
cmax = max(float(curve.max()), 1e-3)
|
||
cmin = min(float(curve.min()), 0.0)
|
||
for j in range(1, len(curve)):
|
||
x0 = int((j - 1) / (len(curve) - 1) * 460) + 10
|
||
x1 = int(j / (len(curve) - 1) * 460) + 10
|
||
y0 = int(150 - (curve[j - 1] - cmin) / (cmax - cmin) * 130)
|
||
y1 = int(150 - (curve[j] - cmin) / (cmax - cmin) * 130)
|
||
cv2.line(img, (x0, y0), (x1, y1), (255, 0, 0), 1)
|
||
xi = int((0.0 - ANGLES[0]) / (ANGLES[-1] - ANGLES[0]) * 460) + 10
|
||
cv2.line(img, (xi, 10), (xi, 150), (0, 200, 0), 1)
|
||
cv2.putText(img, f"{name[:12]} scale={sc:.2f}", (10, 14),
|
||
cv2.FONT_HERSHEY_SIMPLEX, 0.4, (0, 0, 0), 1)
|
||
cv2.imwrite(str(OUT / f"{name[:12]}-rot.png"), img)
|
||
except Exception as e:
|
||
print(f"{name[:12]} 处理失败: {e}")
|
||
lines.append(f"{name[:12]}\tERROR\t{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")
|
||
n_ok = sum(1 for l in lines if "\t目标\t" in l)
|
||
print(f"\n{n_ok}/{len(pairs)} 对定位到零旋转目标；结果已写入 {OUT}/")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|