356f7aa502
Keeps the model's uncertainty out of the UI: decide_identity() separates raw matcher scores into accepted / ambiguous / unknown / rejected, and only an accepted result can name a SKU. Zero evidence is rejected, never a product. observe.py adds frame-local observations and one-zone scan sessions (draft, finalize, cancel, rescan) with replay, staleness and backpressure rules. pipeline.process_click() now reports the identity level instead of failing when the match is not accepted. Incorporates earlier uncommitted session work in this repository (identity.py, observe.py, yolo_adapter.py, test_observe.py, and the pipeline hunk those needed); yolo_adapter keeps its decoded frame per thread so concurrent sessions cannot swap images. Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
106 lines
3.5 KiB
Python
106 lines
3.5 KiB
Python
"""Identity ranking for catalog scores.
|
|
|
|
Separates raw matcher scores from accept/unknown/ambiguous/reject so a
|
|
zero-score crop cannot become a SKU. Thresholds are conservative defaults
|
|
for the 21-SKU closed set; they are not calibrated accuracy claims.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, List, Optional, Sequence, Tuple
|
|
|
|
IDENTITY_ACCEPTED = "accepted"
|
|
IDENTITY_AMBIGUOUS = "ambiguous"
|
|
IDENTITY_UNKNOWN = "unknown"
|
|
IDENTITY_REJECTED = "rejected"
|
|
|
|
# ponytail: closed-set defaults. Recalibrate when held-out scores exist.
|
|
DEFAULT_MIN_SCORE = 1.0
|
|
DEFAULT_MIN_MARGIN = 0.15
|
|
DEFAULT_MIN_CONFIDENCE = 0.55
|
|
|
|
|
|
def relative_confidence(top_score: float, runner_up_score: float) -> float:
|
|
"""Existing matcher formula: top / (top + 0.5 * runner)."""
|
|
return round(min(1.0, top_score / max(1.0, (top_score + runner_up_score * 0.5))), 4)
|
|
|
|
|
|
def decide_identity(
|
|
scores: Dict[str, float],
|
|
*,
|
|
min_score: float = DEFAULT_MIN_SCORE,
|
|
min_margin: float = DEFAULT_MIN_MARGIN,
|
|
min_confidence: float = DEFAULT_MIN_CONFIDENCE,
|
|
details: Optional[Dict[str, Any]] = None,
|
|
method: str = "hybrid",
|
|
) -> Dict[str, Any]:
|
|
"""Rank catalog scores into accepted / ambiguous / unknown / rejected.
|
|
|
|
Zero or missing evidence is rejected. A weak but unique top score is
|
|
unknown. Two close scores are ambiguous. Only a unique, above-threshold
|
|
score is accepted.
|
|
"""
|
|
if not scores:
|
|
return {
|
|
"status": IDENTITY_REJECTED,
|
|
"method": method,
|
|
"best_sku": None,
|
|
"score": 0.0,
|
|
"confidence": 0.0,
|
|
"identity_level": IDENTITY_REJECTED,
|
|
"ranked_candidates": [],
|
|
"details": {},
|
|
"message": "No templates available",
|
|
}
|
|
|
|
sorted_candidates: List[Tuple[str, float]] = sorted(
|
|
scores.items(), key=lambda x: x[1], reverse=True
|
|
)
|
|
top_sku, top_score = sorted_candidates[0]
|
|
runner_up_sku, runner_up_score = (
|
|
sorted_candidates[1] if len(sorted_candidates) > 1 else (None, 0.0)
|
|
)
|
|
confidence = relative_confidence(top_score, runner_up_score)
|
|
ranked = [{"sku_id": s, "score": round(sc, 2)} for s, sc in sorted_candidates[:5]]
|
|
top_details = (details or {}).get(top_sku, {})
|
|
|
|
if top_score <= 0.0:
|
|
level = IDENTITY_REJECTED
|
|
message = "No matching evidence"
|
|
best = None
|
|
elif (
|
|
runner_up_sku is not None
|
|
and top_score > 0.0
|
|
and (top_score - runner_up_score) < min_margin * max(top_score, 1.0)
|
|
):
|
|
level = IDENTITY_AMBIGUOUS
|
|
message = f"Close scores: {top_sku} vs {runner_up_sku}"
|
|
best = None
|
|
elif top_score < min_score or confidence < min_confidence:
|
|
level = IDENTITY_UNKNOWN
|
|
message = "Top score below accept threshold"
|
|
best = None
|
|
else:
|
|
level = IDENTITY_ACCEPTED
|
|
message = "Accepted unique catalog match"
|
|
best = top_sku
|
|
|
|
return {
|
|
"status": "success" if level == IDENTITY_ACCEPTED else level,
|
|
"method": method,
|
|
"best_sku": best,
|
|
"score": round(top_score, 2),
|
|
"confidence": confidence,
|
|
"identity_level": level,
|
|
"ranked_candidates": ranked,
|
|
"details": top_details,
|
|
"message": message,
|
|
}
|
|
|
|
|
|
def identity_from_ranked(
|
|
ranked_candidates: Sequence[Dict[str, Any]],
|
|
**kwargs: Any,
|
|
) -> Dict[str, Any]:
|
|
scores = {str(row["sku_id"]): float(row["score"]) for row in ranked_candidates}
|
|
return decide_identity(scores, **kwargs)
|