product-recognition-demo/recognization/code/identity.py

"""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)


# The gallery's scale is a cosine similarity, where same-product pairs sit above 0.6
# and different products below 0.5 - not an inlier count, which has no such meaning.
COSINE_ACCEPT = 0.60
COSINE_AMBIGUOUS = 0.50
COSINE_MARGIN = 0.04


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",
    scale: str = "inlier",
) -> 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
    )
    if scale == "cosine":
        # Thresholds that mean something on a similarity scale.
        top_sku, top_score = sorted_candidates[0]
        runner_score = sorted_candidates[1][1] if len(sorted_candidates) > 1 else 0.0
        margin = top_score - runner_score
        if top_score >= COSINE_ACCEPT and margin >= COSINE_MARGIN:
            level = IDENTITY_ACCEPTED
        elif top_score >= COSINE_AMBIGUOUS:
            level = IDENTITY_AMBIGUOUS
        else:
            level = IDENTITY_REJECTED
        return {
            "status": level,
            "method": method,
            "best_sku": top_sku if level == IDENTITY_ACCEPTED else None,
            "score": float(top_score),
            "confidence": float(top_score),
            "identity_level": level,
            "ranked_candidates": [
                {"sku_id": sku, "score": float(value)} for sku, value in sorted_candidates[:3]
            ],
            "details": details or {},
            "message": None,
        }
    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)