ca3882c59c
Nuovo helper _self_validate(): post-stima parametri, esegue dry-run training+find sul template stesso e regola i parametri se subottimali. Loop di auto-correzione (analogo a Halcon inspect_shape_model): 1. Se top-level piramide ha <8 feature → riduce pyramid_levels 2. Se train produce 0 varianti → dimezza weak/strong_grad 3. Se find sul template fallisce → riduce soglie + num_features 4. Se self-score < 0.7 → abbassa weak_grad Costo: 1 train minimale (1 variante) + 1 find su canvas tpl + padding, ~50ms su template 100x100. Ne vale la pena per evitare match-time errors su scene reali con parametri estimato male. Esposto via auto_tune(self_validate=True) default; meta '_self_score' e '_validation' nel dict risultato per logging UI. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
394 lines
15 KiB
Python
394 lines
15 KiB
Python
"""Auto-tune parametri PM2D da analisi del template.
|
|
|
|
Analizza la ROI del modello e suggerisce valori ragionevoli per i principali
|
|
parametri del `LineShapeMatcher`, tenendo conto di:
|
|
|
|
- **distribuzione magnitude del gradiente** → soglie `weak_grad` / `strong_grad`
|
|
- **numero di edge utili** → `num_features`
|
|
- **dimensione template** → `pyramid_levels`, `spread_radius`
|
|
- **simmetria rotazionale** (autocorrelazione su rotazione) → `angle_range_deg`
|
|
- **entropia orientamenti** → suggerimento `min_score`
|
|
|
|
Ritorna dict con i key esatti del form `edit_params`.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
from collections import OrderedDict
|
|
|
|
import cv2
|
|
import numpy as np
|
|
|
|
|
|
def _to_gray(img: np.ndarray) -> np.ndarray:
|
|
if img.ndim == 3:
|
|
return cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
|
|
return img
|
|
|
|
|
|
# Cache in-memory (LRU) dei risultati auto_tune per stesso input ROI.
|
|
_TUNE_CACHE: OrderedDict[str, dict] = OrderedDict()
|
|
_TUNE_CACHE_SIZE = 32
|
|
|
|
|
|
def detect_rotational_symmetry(
|
|
gray: np.ndarray, step_deg: float = 5.0, corr_thresh: float = 0.75,
|
|
) -> dict:
|
|
"""Rileva simmetria rotazionale su edge map (più robusto a sfondo uniforme).
|
|
|
|
Downsample a max 128 px prima di correlare per abbattere il costo
|
|
O(n_angles · H · W) senza perdere precisione (la simmetria rotazionale
|
|
è invariante a subsampling moderato).
|
|
|
|
Ritorna dict con:
|
|
- order: int, 1=nessuna, 2=180°, 3=120°, 4=90°, 6=60°, 8=45°
|
|
- period_deg: float, periodo minimo di simmetria (360/order)
|
|
- confidence: float [0..1], correlazione minima tra rotazioni equivalenti
|
|
"""
|
|
h, w = gray.shape
|
|
target = 128
|
|
if max(h, w) > target:
|
|
sf = target / max(h, w)
|
|
new_w = max(32, int(w * sf))
|
|
new_h = max(32, int(h * sf))
|
|
gray = cv2.resize(gray, (new_w, new_h), interpolation=cv2.INTER_AREA)
|
|
h, w = gray.shape
|
|
# Usa magnitude gradiente (rotation-invariant rispetto a bg uniforme)
|
|
gx = cv2.Sobel(gray, cv2.CV_32F, 1, 0, ksize=3)
|
|
gy = cv2.Sobel(gray, cv2.CV_32F, 0, 1, ksize=3)
|
|
mag = cv2.magnitude(gx, gy).astype(np.float32)
|
|
|
|
center = (w / 2.0, h / 2.0)
|
|
ref = mag
|
|
|
|
correlations: list[tuple[float, float]] = []
|
|
for ang in np.arange(step_deg, 360.0, step_deg):
|
|
M = cv2.getRotationMatrix2D(center, float(ang), 1.0)
|
|
rot = cv2.warpAffine(
|
|
mag, M, (w, h), borderValue=0.0,
|
|
)
|
|
rm = ref - ref.mean()
|
|
rs = rot - rot.mean()
|
|
denom = np.sqrt((rm * rm).sum() * (rs * rs).sum()) + 1e-9
|
|
c = float((rm * rs).sum() / denom)
|
|
correlations.append((float(ang), c))
|
|
|
|
# Candidati simmetria: 2,3,4,6,8 (90/45)
|
|
candidates = [2, 3, 4, 6, 8]
|
|
best_order = 1
|
|
best_conf = 0.0
|
|
for order in candidates:
|
|
period = 360.0 / order
|
|
# Verifica che ALLE rotazioni n*period (n=1..order-1) ci sia alta corr
|
|
corrs = []
|
|
for n in range(1, order):
|
|
target = period * n
|
|
# trova angolo più vicino in correlations
|
|
closest = min(correlations, key=lambda p: abs(p[0] - target))
|
|
if abs(closest[0] - target) > step_deg * 1.5:
|
|
corrs.append(0.0)
|
|
else:
|
|
corrs.append(closest[1])
|
|
conf = min(corrs) if corrs else 0.0
|
|
if conf >= corr_thresh and conf > best_conf:
|
|
best_order = order
|
|
best_conf = conf
|
|
return {
|
|
"order": best_order,
|
|
"period_deg": 360.0 / best_order,
|
|
"confidence": best_conf,
|
|
}
|
|
|
|
|
|
def analyze_gradients(gray: np.ndarray) -> dict:
|
|
"""Statistiche magnitude / orientation gradiente."""
|
|
gx = cv2.Sobel(gray, cv2.CV_32F, 1, 0, ksize=3)
|
|
gy = cv2.Sobel(gray, cv2.CV_32F, 0, 1, ksize=3)
|
|
mag = cv2.magnitude(gx, gy)
|
|
|
|
# Percentili magnitude: p55/p85 usati per soglie weak/strong (più aderenti
|
|
# alla distribuzione reale rispetto a p50/p80 + clamp).
|
|
p50 = float(np.percentile(mag, 50))
|
|
p55 = float(np.percentile(mag, 55))
|
|
p80 = float(np.percentile(mag, 80))
|
|
p85 = float(np.percentile(mag, 85))
|
|
p95 = float(np.percentile(mag, 95))
|
|
mag_max = float(mag.max())
|
|
|
|
# Numero pixel "forti"
|
|
strong_pct = float((mag > p95).sum()) / mag.size
|
|
weak_pct = float((mag > p50).sum()) / mag.size
|
|
|
|
# Entropia orientamenti (solo pixel forti)
|
|
ang = np.arctan2(gy, gx)
|
|
ang_mod = np.where(ang < 0, ang + np.pi, ang)
|
|
mask = mag > p80
|
|
if mask.sum() > 10:
|
|
bins_count, _ = np.histogram(
|
|
ang_mod[mask], bins=16, range=(0, np.pi),
|
|
)
|
|
p = bins_count / (bins_count.sum() + 1e-9)
|
|
ent = float(-np.sum(p * np.log(p + 1e-9)) / np.log(16))
|
|
else:
|
|
ent = 0.0
|
|
|
|
return {
|
|
"p50": p50, "p55": p55, "p80": p80, "p85": p85, "p95": p95,
|
|
"mag_max": mag_max,
|
|
"strong_pct": strong_pct, "weak_pct": weak_pct,
|
|
"orient_entropy": ent,
|
|
"n_pixels": mag.size,
|
|
"n_strong": int((mag > p95).sum()),
|
|
}
|
|
|
|
|
|
def _cache_key(template_bgr: np.ndarray, mask: np.ndarray | None) -> str:
|
|
h = hashlib.md5()
|
|
h.update(np.ascontiguousarray(template_bgr).tobytes())
|
|
h.update(f"shape={template_bgr.shape}".encode())
|
|
if mask is not None:
|
|
h.update(np.ascontiguousarray(mask).tobytes())
|
|
return h.hexdigest()
|
|
|
|
|
|
def _self_validate(template_bgr: np.ndarray, params: dict,
|
|
mask: np.ndarray | None = None) -> dict:
|
|
"""Halcon-style self-validation: train il matcher coi parametri tentativi
|
|
e verifica che il template stesso sia trovato con recall ≥ 1.0.
|
|
|
|
Se recall < target o score basso, regola i parametri:
|
|
- alza weak_grad se troppi edge spuri (recall solido ma molti picchi falsi)
|
|
- abbassa strong_grad se troppe feature scartate (low feature count)
|
|
- riduce pyramid_levels se variants[0].levels[top] ha <8 feature
|
|
|
|
Halcon usa internamente questo loop in inspect_shape_model. Costo: 1
|
|
train + 1 find sul template (~50ms su template 100x100). Ne vale la
|
|
pena se evita match-time errors su scene reali.
|
|
|
|
Mutates `params` in place e ritorna lo stesso dict per chaining.
|
|
"""
|
|
# Import lazy: evita ciclo (line_matcher importa nulla da auto_tune)
|
|
from pm2d.line_matcher import LineShapeMatcher
|
|
|
|
# Caso degenerato: troppe poche feature pre-validation → riduci soglia
|
|
if params.get("_n_strong_pixels", 0) < 30:
|
|
params["weak_grad"] = max(15.0, params["weak_grad"] * 0.6)
|
|
params["strong_grad"] = max(30.0, params["strong_grad"] * 0.6)
|
|
|
|
# Train minimale: 1 sola pose orientazione 0 (range degenerato che
|
|
# produce comunque 1 variante via fallback in _angle_list).
|
|
m = LineShapeMatcher(
|
|
num_features=params["num_features"],
|
|
weak_grad=params["weak_grad"],
|
|
strong_grad=params["strong_grad"],
|
|
angle_range_deg=(0.0, 0.0), # fallback _angle_list = [0.0]
|
|
angle_step_deg=10.0,
|
|
scale_range=(1.0, 1.0),
|
|
spread_radius=params["spread_radius"],
|
|
pyramid_levels=params["pyramid_levels"],
|
|
)
|
|
n_var = m.train(template_bgr, mask=mask)
|
|
if n_var == 0:
|
|
# Soglie troppo alte: nessuna variante generata → dimezza
|
|
params["weak_grad"] = max(15.0, params["weak_grad"] * 0.5)
|
|
params["strong_grad"] = max(30.0, params["strong_grad"] * 0.5)
|
|
params["_validation"] = "fallback: soglie dimezzate (no variants)"
|
|
return params
|
|
|
|
# Verifica densita' feature al top-level (rischio collasso)
|
|
top_lvl = m.variants[0].levels[-1]
|
|
if top_lvl.n < 8 and params["pyramid_levels"] > 1:
|
|
params["pyramid_levels"] = max(1, params["pyramid_levels"] - 1)
|
|
params["_validation"] = (
|
|
f"pyramid_levels ridotto a {params['pyramid_levels']} "
|
|
f"(top aveva {top_lvl.n} feature)"
|
|
)
|
|
return params
|
|
|
|
# Self-find: cerca il template stesso nella propria immagine
|
|
h, w = template_bgr.shape[:2]
|
|
# Embed template in scena leggermente più grande per evitare bordo
|
|
pad = 20
|
|
canvas = np.full(
|
|
(h + 2 * pad, w + 2 * pad, 3 if template_bgr.ndim == 3 else 1),
|
|
128, dtype=np.uint8,
|
|
)
|
|
canvas[pad:pad + h, pad:pad + w] = template_bgr
|
|
matches = m.find(
|
|
canvas, min_score=0.3, max_matches=5,
|
|
verify_ncc=False, # template stesso → NCC = 1 sempre, skip per velocita'
|
|
refine_angle=False, subpixel=False,
|
|
nms_iou_threshold=0.3,
|
|
)
|
|
if not matches:
|
|
# Nessun match sul proprio template: parametri troppo restrittivi
|
|
params["weak_grad"] = max(15.0, params["weak_grad"] * 0.7)
|
|
params["strong_grad"] = max(30.0, params["strong_grad"] * 0.7)
|
|
params["num_features"] = max(48, int(params["num_features"] * 0.8))
|
|
params["_validation"] = "soglie/feature ridotte (no self-match)"
|
|
return params
|
|
|
|
# Misura score top match
|
|
top_score = float(matches[0].score)
|
|
params["_self_score"] = round(top_score, 3)
|
|
if top_score < 0.7:
|
|
# Score basso sul template stesso = parametri davvero subottimali
|
|
params["weak_grad"] = max(15.0, params["weak_grad"] * 0.85)
|
|
params["_validation"] = (
|
|
f"weak_grad ridotto (self-score era {top_score:.2f})"
|
|
)
|
|
else:
|
|
params["_validation"] = f"OK (self-score {top_score:.2f})"
|
|
return params
|
|
|
|
|
|
def auto_tune(
|
|
template_bgr: np.ndarray,
|
|
mask: np.ndarray | None = None,
|
|
angle_tolerance_deg: float | None = None,
|
|
angle_center_deg: float = 0.0,
|
|
self_validate: bool = True,
|
|
) -> dict:
|
|
"""Analizza template e ritorna dict parametri suggeriti.
|
|
|
|
Chiavi compatibili con edit_params PARAM_SCHEMA.
|
|
|
|
angle_tolerance_deg: se != None, restringe angle_range a
|
|
(center - tol, center + tol). Usare quando l'orientamento del
|
|
pezzo e' noto a priori (feeder con guida, posizionamento
|
|
meccanico): training molto piu rapido (24x meno varianti per
|
|
tol=15° vs 360° pieno).
|
|
|
|
self_validate: se True (default), dopo la stima dei parametri
|
|
esegue un dry-run del matching sul template stesso e regola
|
|
weak_grad/strong_grad/pyramid_levels se i parametri tentativi
|
|
non garantiscono auto-match (Halcon-style inspect_shape_model).
|
|
|
|
Risultato cachato in-memory (LRU): ri-chiamare con stessa ROI è O(1).
|
|
"""
|
|
ck = _cache_key(template_bgr, mask)
|
|
if angle_tolerance_deg is not None:
|
|
ck = f"{ck}|tol={angle_tolerance_deg}|c={angle_center_deg}"
|
|
cached = _TUNE_CACHE.get(ck)
|
|
if cached is not None:
|
|
_TUNE_CACHE.move_to_end(ck)
|
|
return dict(cached)
|
|
|
|
gray = _to_gray(template_bgr)
|
|
h, w = gray.shape
|
|
if mask is not None:
|
|
# Zero fuori maschera per statistiche
|
|
gray_for_stats = np.where(mask > 0, gray, int(np.median(gray))).astype(np.uint8)
|
|
else:
|
|
gray_for_stats = gray
|
|
|
|
stats = analyze_gradients(gray_for_stats)
|
|
sym = detect_rotational_symmetry(gray_for_stats)
|
|
|
|
# Soglie magnitude: usa percentili reali (p85/p55) senza clamp duro a 100.
|
|
# Sobel ksize=3 su uint8 può arrivare a ~1020, quindi clamp massimo 400
|
|
# evita saturazione del threshold su template ad alto contrasto.
|
|
strong_grad = float(np.clip(stats["p85"], 30.0, 400.0))
|
|
weak_grad = float(np.clip(stats["p55"], 15.0, strong_grad * 0.7))
|
|
|
|
# num_features: ibrido perimetro + densità. Target = min(perimeter_budget,
|
|
# density_budget) per non generare più feature di quante edge nitide siano
|
|
# disponibili, ma neanche meno di quante il perimetro possa tracciare.
|
|
perim_budget = int(2 * (h + w) * 0.4) # ~40% dei pixel di perimetro
|
|
density_budget = int(stats["n_strong"] / 20) # 1 feature ogni ~20 px forti
|
|
target_feat = int(np.clip(min(perim_budget, density_budget), 64, 192))
|
|
|
|
# pyramid_levels in base a dimensione minima E densità feature: un template
|
|
# grande ma povero di feature non deve scendere troppi livelli (rischio
|
|
# collasso a <16 feature al top level).
|
|
min_side = min(h, w)
|
|
if min_side < 60:
|
|
pyr = 1
|
|
elif min_side < 120:
|
|
pyr = 2
|
|
elif min_side < 320:
|
|
pyr = 3
|
|
else:
|
|
pyr = 4
|
|
# Cap: non scendere sotto ~16 feature al top level (feature ÷ 4^(pyr-1))
|
|
max_pyr_from_feat = max(1, int(np.floor(np.log2(max(1, target_feat / 16.0)) / 2.0)) + 1)
|
|
pyr = min(pyr, max_pyr_from_feat)
|
|
|
|
# spread_radius proporzionale a risoluzione + pyramid (tolleranza ~1% dim)
|
|
spread_radius = int(np.clip(max(3, min_side * 0.02), 3, 8))
|
|
|
|
# angle range: priorita' a tolerance hint utente, poi simmetria rotazionale.
|
|
if angle_tolerance_deg is not None:
|
|
angle_min = float(angle_center_deg - angle_tolerance_deg)
|
|
angle_max = float(angle_center_deg + angle_tolerance_deg)
|
|
else:
|
|
angle_min = 0.0
|
|
angle_max = 360.0 / sym["order"] if sym["order"] > 1 else 360.0
|
|
|
|
# min_score: se entropia orient alta → template distintivo → soglia alta ok
|
|
# se entropia bassa → template ambiguo → soglia più permissiva
|
|
if stats["orient_entropy"] > 0.75:
|
|
min_score = 0.65
|
|
elif stats["orient_entropy"] > 0.55:
|
|
min_score = 0.55
|
|
else:
|
|
min_score = 0.45
|
|
|
|
# angle step adattivo (Halcon-style): atan(2/max_side) deg, clampato.
|
|
# Template grande → step fine (rotazione minima visibile su perimetro).
|
|
# Template piccolo → step grosso (over-sampling = sprecato).
|
|
max_side = max(h, w)
|
|
angle_step = float(np.clip(np.degrees(np.arctan2(2.0, max_side)), 1.0, 8.0))
|
|
|
|
result = {
|
|
"backend": "line",
|
|
"angle_min": angle_min,
|
|
"angle_max": angle_max,
|
|
"angle_step": angle_step,
|
|
"scale_min": 1.0,
|
|
"scale_max": 1.0,
|
|
"scale_step": 0.1,
|
|
"min_score": round(min_score, 2),
|
|
"max_matches": 25,
|
|
"nms_radius": 0,
|
|
"num_features": target_feat,
|
|
"weak_grad": round(weak_grad, 1),
|
|
"strong_grad": round(strong_grad, 1),
|
|
"spread_radius": spread_radius,
|
|
"pyramid_levels": pyr,
|
|
"verify_threshold": 0.4,
|
|
# meta (non in PARAM_SCHEMA, usato per log)
|
|
"_symmetry_order": sym["order"],
|
|
"_symmetry_conf": round(sym["confidence"], 2),
|
|
"_orient_entropy": round(stats["orient_entropy"], 2),
|
|
"_n_strong_pixels": stats["n_strong"],
|
|
}
|
|
# Halcon-style self-validation: dry-run training+find sul template per
|
|
# auto-correggere parametri tentativi che non garantirebbero match.
|
|
if self_validate:
|
|
result = _self_validate(template_bgr, result, mask=mask)
|
|
# Round numerici dopo eventuali aggiustamenti
|
|
result["weak_grad"] = round(result["weak_grad"], 1)
|
|
result["strong_grad"] = round(result["strong_grad"], 1)
|
|
# Store in LRU cache
|
|
_TUNE_CACHE[ck] = dict(result)
|
|
_TUNE_CACHE.move_to_end(ck)
|
|
while len(_TUNE_CACHE) > _TUNE_CACHE_SIZE:
|
|
_TUNE_CACHE.popitem(last=False)
|
|
return result
|
|
|
|
|
|
def summarize(tune: dict) -> str:
|
|
"""Stringa one-line delle scelte principali."""
|
|
so = tune.get("_symmetry_order", 1)
|
|
sc = tune.get("_symmetry_conf", 0)
|
|
ent = tune.get("_orient_entropy", 0)
|
|
return (
|
|
f"sym={so}x (conf={sc:.2f}) entropia={ent:.2f} "
|
|
f"feat={tune['num_features']} pyr={tune['pyramid_levels']} "
|
|
f"grad={tune['weak_grad']:.0f}/{tune['strong_grad']:.0f} "
|
|
f"ang=[0..{tune['angle_max']:.0f}]@{tune['angle_step']:.0f}d "
|
|
f"min_score={tune['min_score']}"
|
|
)
|