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. Three score scales are recognized:

- `inlier`: Van's composite, kept as the selectable baseline lane (E-1).
- `local`: `identity_lanes.LocalLane`'s margin-over-runner-up, in inlier units.
- `cosine`: the embedding gallery, fenced off and not in the live path.

The `local` line is measured on the corrected image6 labels: 2.0 keeps 95% of
shown labels correct at 59% coverage with the text lane, 1.0 keeps 87% at 66%.
It is tuned on one photo, so it is a direction, not a calibration; E-4 replaces
it once Tier-2 held-out labels exist.
"""

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 for the baseline inlier scale. Recalibrate when
# held-out scores exist.
DEFAULT_MIN_SCORE = 1.0
DEFAULT_MIN_MARGIN = 0.15
DEFAULT_MIN_CONFIDENCE = 0.55

# `local` scale (identity_lanes.LocalLane). The value is already a margin over
# the runner-up, so the relative-margin test does not apply to it: a single
# absolute line is the whole gate.
LOCAL_ACCEPT = 2.0


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)


# Cosine scale (gallery). Same-product pairs were measured above 0.6 and
# different products below 0.5. These are not a new calibration.
COSINE_ACCEPT = 0.60
COSINE_AMBIGUOUS = 0.50
COSINE_MARGIN = 0.04
VOTE_CAP = 3


def _finite(value: Any) -> Optional[float]:
    try:
        number = float(value)
    except (TypeError, ValueError):
        return None
    if number != number or number in (float("inf"), float("-inf")):
        return None
    return number


def aggregate_votes(
    observations: Sequence[Dict[str, float]],
    *,
    cap: int = VOTE_CAP,
) -> Dict[str, float]:
    """Mean of the newest observations, at most `cap`.

    Older frames are dropped, not averaged in. An empty observation still
    counts, so a later miss can outvote an earlier weak hit. Inlier and
    cosine scores must not be mixed in one list.
    """
    window = list(observations)[-max(1, int(cap)):]
    if not window:
        return {}
    totals: Dict[str, float] = {}
    for scores in window:
        for sku_id, score in (scores or {}).items():
            number = _finite(score)
            if number is None:
                continue
            totals[str(sku_id)] = totals.get(str(sku_id), 0.0) + number
    count = len(window)
    return {sku_id: total / count for sku_id, total in totals.items()}


def _cosine_decision(
    sorted_candidates: List[Tuple[str, float]],
    *,
    method: str,
    details: Optional[Dict[str, Any]],
    accept: float,
    ambiguous: float,
    margin_min: float,
) -> Dict[str, Any]:
    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 >= accept and margin >= margin_min:
        level = IDENTITY_ACCEPTED
    elif top_score >= 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,
    }


def decision_for_match(
    match: Any,
    *,
    method: str = "gallery",
    margin: float = COSINE_MARGIN,
) -> Dict[str, Any]:
    """Rank one gallery match. Rejection is the gallery's, not a second threshold.

    `match()` already applied the raw 0.55 line or the head's 0.40 line. Re-ranking
    those scores against COSINE_ACCEPT (0.60) would refuse a head match the gallery
    accepted. The only extra gate here is the published cosine margin.
    """
    similarity = _finite(getattr(match, "similarity", 0.0)) or 0.0
    if match is None or getattr(match, "rejected", False) or not getattr(match, "sku_id", None):
        return {
            "status": IDENTITY_REJECTED,
            "method": method,
            "best_sku": None,
            "score": similarity,
            "confidence": similarity,
            "identity_level": IDENTITY_REJECTED,
            "ranked_candidates": [],
            "details": {},
            "message": "no enrolled product is close enough",
        }
    runner = getattr(match, "runner_up", None)
    runner_score = _finite(getattr(match, "runner_up_similarity", 0.0)) or 0.0
    gap = similarity - runner_score
    close = bool(runner) and gap < margin
    level = IDENTITY_AMBIGUOUS if close else IDENTITY_ACCEPTED
    ranked = [{"sku_id": str(match.sku_id), "score": similarity}]
    if runner:
        ranked.append({"sku_id": str(runner), "score": runner_score})
    return {
        "status": level,
        "method": method,
        "best_sku": None if close else str(match.sku_id),
        "score": similarity,
        "confidence": similarity,
        "identity_level": level,
        "ranked_candidates": ranked,
        "details": {},
        "message": None,
    }


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 scale == "cosine":
        if min_score == DEFAULT_MIN_SCORE:
            min_score = COSINE_ACCEPT
        if min_margin == DEFAULT_MIN_MARGIN:
            min_margin = COSINE_MARGIN
        if min_confidence == DEFAULT_MIN_CONFIDENCE:
            min_confidence = COSINE_AMBIGUOUS
    elif scale == "local" and min_score == DEFAULT_MIN_SCORE:
        min_score = LOCAL_ACCEPT
    cleaned: Dict[str, float] = {}
    for sku_id, score in (scores or {}).items():
        number = _finite(score)
        if number is None:
            continue
        cleaned[str(sku_id)] = number
    if not cleaned:
        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(
        cleaned.items(), key=lambda item: item[1], reverse=True
    )
    if scale == "cosine":
        return _cosine_decision(
            sorted_candidates,
            method=method,
            details=details,
            accept=min_score,
            ambiguous=min_confidence,
            margin_min=min_margin,
        )
    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 scale == "local":
        # The score is already a margin over the runner-up, so the relative
        # margin test below would apply the same correction twice. One absolute
        # line is the whole gate; a candidate under it is unknown, not ambiguous,
        # because the lane has already said which one it prefers.
        if top_score <= 0.0:
            level, message, best = IDENTITY_REJECTED, "No matching evidence", None
        elif top_score < min_score:
            level, message, best = IDENTITY_UNKNOWN, "Top score below accept threshold", None
        else:
            level, message, best = IDENTITY_ACCEPTED, "Accepted unique catalog match", top_sku
        return {
            "status": "success" if level == IDENTITY_ACCEPTED else level,
            "method": method,
            "best_sku": best,
            "score": round(top_score, 2),
            "confidence": round(top_score, 4),
            "identity_level": level,
            "ranked_candidates": ranked,
            "details": top_details,
            "message": message,
        }

    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)