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

"""Score a streamed clip's labels against ground truth: the G1 number.

The stream path was reported as "53 of 131 tracks", which is not comparable to
Van's 57% of labelled facings - the two count different things. But the clip is
built from `add/image6.jpg`, which has labels, so matching each confirmed
track's `label_box` to the labelled facings gives stream-path coverage directly.

    python code/score_stream.py TRACKS.json --labels eval/labels/image6.json

`TRACKS.json` is what `stream_client.py --tracks` writes: one row per track that
settled on a label, with the box that label was measured on, already mapped back
to source-image coordinates.

Reported per run: correct / in-catalog facings, precision, unknown accepts, and
tracks that matched no labelled facing. A track whose box overlaps a facing the
catalog does not enrol is an unenrolled false accept - the dangerous kind, and
the one the evaluator's own precision denominator used to hide.
"""

from __future__ import annotations

import argparse
import json
import os
import sys
from typing import Any, Dict, List, Optional, Sequence, Tuple

current_dir = os.path.dirname(os.path.abspath(__file__))
if current_dir not in sys.path:
    sys.path.insert(0, current_dir)

# A facing clipped by the window edge is not something a run could have named,
# so the denominator keeps only facings at least this far inside the window.
VISIBLE_MIN = 0.5


def iou(a: Sequence[int], b: Sequence[int]) -> float:
    ax1, ay1, ax2, ay2 = a
    bx1, by1, bx2, by2 = b
    inter_w = max(0, min(ax2, bx2) - max(ax1, bx1))
    inter_h = max(0, min(ay2, by2) - max(ay1, by1))
    inter = inter_w * inter_h
    area_a = max(0, ax2 - ax1) * max(0, ay2 - ay1)
    area_b = max(0, bx2 - bx1) * max(0, by2 - by1)
    union = area_a + area_b - inter
    return inter / union if union > 0 else 0.0


def load_facings(labels_path: str) -> List[Dict[str, Any]]:
    with open(labels_path) as handle:
        payload = json.load(handle)
    return payload["facings"]


def enrolled_skus(stock_dir: str) -> set:
    """Canonical SKUs with at least one enrolled view."""
    from rebuild_catalog import sku_of

    return {
        sku_of(name) for name in os.listdir(stock_dir)
        if name.lower().endswith((".jpg", ".jpeg", ".png"))
    }


def visible_fraction(box: Sequence[int], window: Sequence[int]) -> float:
    """Share of a facing's area inside the window the client actually showed."""
    wx0, wy0, wx1, wy1 = window
    bx0, by0, bx1, by1 = box
    inter_w = max(0, min(bx1, wx1) - max(bx0, wx0))
    inter_h = max(0, min(by1, wy1) - max(by0, wy0))
    area = max(0, bx1 - bx0) * max(0, by1 - by0)
    return (inter_w * inter_h) / area if area > 0 else 0.0


def score(tracks: Sequence[Dict[str, Any]], facings: Sequence[Dict[str, Any]],
          stock_dir: str, iou_min: float,
          windows: Optional[Sequence[Sequence[int]]] = None
          ) -> Dict[str, Any]:
    """Score the labels a stream settled on.

    `windows` are the source-image rects the client showed, as
    [x0, y0, x1, y1] each. The denominator is the facings at least half inside
    one of them, in x and y both: a 1280x720 window over a 1500x2000 photo sees
    24 of the 63 in-catalog facings, and a facing clipped by the window edge is
    not something any run could have named. A panned run shows a different rect
    each frame, so the reachable set is the union. Without it the denominator is
    every in-catalog facing.

    A correct answer is counted before the unenrolled and unknown branches. The
    text lane names SKUs the image lane has no view for, which is the point of
    it, and scoring that as a false accept punishes the lane for working.
    """
    if windows:
        facings = [
            f for f in facings
            if max(visible_fraction(f["box"], w) for w in windows) >= VISIBLE_MIN
        ]
    in_catalog = [f for f in facings if f["label"] != "unknown"]
    enrolled = enrolled_skus(stock_dir)

    correct = 0
    wrong = 0
    unenrolled_false = 0
    unknown_false = 0
    unmatched = 0
    rows: List[Dict[str, Any]] = []
    # Coverage is per facing, precision is per label. A track that restarts on
    # the same product, or two tracks on one facing, is two claims but one
    # facing: counting tracks as coverage reports 138% of a 24-facing window.
    named: Dict[int, str] = {}

    for track in tracks:
        box = track.get("label_box") or []
        if len(box) != 4:
            unmatched += 1
            continue
        best, best_iou = None, 0.0
        for facing in facings:
            value = iou(box, facing["box"])
            if value > best_iou:
                best, best_iou = facing, value
        if best is None or best_iou < iou_min:
            unmatched += 1
            rows.append({"track_id": track["track_id"], "predicted": track["sku_id"],
                         "truth": None, "iou": round(best_iou, 3), "outcome": "unmatched"})
            continue
        truth = best["label"]
        predicted = track["sku_id"]
        if predicted == truth:
            correct += 1
            outcome = "correct"
            named[best["index"]] = predicted
        elif truth == "unknown":
            unknown_false += 1
            outcome = "unknown_false_accept"
        elif truth not in enrolled:
            unenrolled_false += 1
            outcome = "unenrolled_false_accept"
        else:
            wrong += 1
            outcome = "wrong"
        rows.append({"track_id": track["track_id"], "predicted": predicted,
                     "truth": truth, "iou": round(best_iou, 3), "outcome": outcome})

    shown = correct + wrong + unenrolled_false + unknown_false
    return {
        "in_catalog_facings": len(in_catalog),
        "facings_named": len(named),
        "tracks_scored": len(tracks),
        "correct": correct,
        "wrong": wrong,
        "unenrolled_false_accepts": unenrolled_false,
        "unknown_false_accepts": unknown_false,
        "unmatched_tracks": unmatched,
        "shown": shown,
        "coverage": round(len(named) / len(in_catalog), 4) if in_catalog else 0.0,
        "precision": round(correct / shown, 4) if shown else 0.0,
        "rows": rows,
    }


def main(argv: Optional[Sequence[str]] = None) -> int:
    parser = argparse.ArgumentParser(description="Score streamed labels against ground truth")
    parser.add_argument("tracks", help="JSON written by stream_client.py --tracks")
    parser.add_argument("--labels", default="eval/labels/image6.json")
    parser.add_argument("--stock-dir", default="stock-lopia-clean")
    parser.add_argument("--iou", type=float, default=0.5)
    parser.add_argument("--json", dest="json_path", default=None)
    parser.add_argument("--window", default=None,
                        help="x0,y0,x1,y1 the client showed; the denominator is "
                             "restricted to facings at least half inside it")
    parser.add_argument("--verbose", action="store_true", help="print every scored track")
    args = parser.parse_args(argv)

    with open(args.tracks) as handle:
        payload = json.load(handle)
    windows = None
    if args.window:
        windows = [tuple(int(part) for part in args.window.split(","))]
    elif payload.get("windows"):
        windows = [tuple(w) for w in payload["windows"]]
    report = score(payload["tracks"], load_facings(args.labels), args.stock_dir,
                   args.iou, windows)
    report["source"] = payload.get("source")
    report["pace"] = payload.get("pace")
    report["seconds"] = payload.get("seconds")
    report["windows"] = [list(w) for w in windows] if windows else None

    print(f"{os.path.basename(args.tracks)}  pace={report['pace']}  "
          f"{report['seconds']}s")
    print(f"  facings named {report['facings_named']}/{report['in_catalog_facings']} "
          f"in-catalog ({report['coverage']:.0%})  labels correct {report['correct']}  "
          f"precision {report['precision']:.0%}  shown {report['shown']}")
    print(f"  wrong {report['wrong']}  unenrolled false accepts "
          f"{report['unenrolled_false_accepts']}  unknown false accepts "
          f"{report['unknown_false_accepts']}  unmatched tracks {report['unmatched_tracks']}")
    if args.verbose:
        for row in report["rows"]:
            print(f"    t{row['track_id']:4d} {row['outcome']:24s} "
                  f"pred={row['predicted']} truth={row['truth']} iou={row['iou']}")

    if args.json_path:
        with open(args.json_path, "w") as handle:
            json.dump(report, handle, indent=2)
    return 0


if __name__ == "__main__":
    sys.exit(main())