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

"""Joint sweep of the two open-set gates: LOCAL_ACCEPT x TEXT_MIN_SHARED.

Both lines were fitted on image6, where every crop is a facing of a product the
shelf actually holds. That set can only measure false accepts on the products it
happens to contain, so it cannot price a line on its own - the far-unknown rate
comes from `data/external-shelves/*`: Wikimedia Commons shelf photos that share
nothing with the catalog, where every accept is a false accept by construction.

Each cell is produced by calling the shipped `DecisionPolicy` with that cell's
values and the catalog signatures, never by re-implementing the gate in the
sweep: a sweep that scores its own copy of the policy reports on a machine the
server does not run. `TEXT_MIN_SHARED` is a floor on the evidence behind the
text lane's *answer*. It is deliberately not a floor on the score - zeroing a
row reorders the ranking, so the margin gate would compare a different pair -
and deliberately not a floor on the veto, which only ever refuses. The constant
in `identity_lanes.py` says the same thing next to the code that does it.

The distractor side is measured the way the phone measures it - a 1280-long-side
frame at `LiveSettings.conf`, the path `code/eval_distractors.py` runs. The
shelf is scored at both resolutions: the full-resolution column is the one the
published table quotes, the live column is what the phone sees.

RootSIFT and OCR dominate the runtime and neither depends on the gates, so the
per-crop evidence is cached. Without `--cache` the sweep rebuilds it every run
and prints the same numbers, only slower.

    .venv/bin/python code/sweep_accept_lines.py \\
        --cache data/evaluations/2026-09-29-session/scratch/accept_line_cache.pkl
"""

from __future__ import annotations

import argparse
import glob
import os
import pickle
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)

import cv2  # noqa: E402

from eval_identity import (  # noqa: E402
    EXCLUDE,
    LIVE_LONG_SIDE,
    UNKNOWN,
    LocalLaneAdapter,
    at_live_resolution,
    crop_of,
    load_labels,
)
from identity_lanes import LOCAL_ACCEPT, TEXT_MIN_SHARED, DecisionPolicy  # noqa: E402

SHELF_LABELS = "eval/labels/image6.json"
SHELF_IMAGE = "add/image6.jpg"
DISTRACTOR_DIRS = (
    "data/external-shelves/wikimedia-commons",
    "data/external-shelves/wikimedia-commons-2",
)
LOCAL_ACCEPTS = (2.0, 2.5, 3.0, 3.5, 4.0, 4.5, 5.0)
TEXT_MIN_SHAREDS = (2, 3, 4, 5, 6)
CACHE_VERSION = 3

# One crop's evidence: the local lane's ranking and the text lane's raw read.
Evidence = Tuple[Any, str]
# One facing: its ground-truth label plus that evidence.
Facing = Tuple[str, Any, str]


def _evidence(lane: LocalLaneAdapter, frame: Any, box: Sequence[int]) -> Optional[Evidence]:
    """Rank and read one crop, or None when the box is off the frame."""
    crop = crop_of(frame, box)
    if crop.size == 0:
        return None
    return lane.rank(crop), lane.text_lane.read(crop)


def build_cache(stock_dir: str) -> Dict[str, Any]:
    """Every crop's evidence, gathered once and swept many times."""
    from engine import EngineSettings, build_default_engine
    from live import LiveSettings

    lane = LocalLaneAdapter(stock_dir)
    image = cv2.imread(SHELF_IMAGE)
    live = at_live_resolution(image)
    factor = live.shape[1] / float(image.shape[1])

    shelf_full: List[Facing] = []
    shelf_live: List[Facing] = []
    for facing in load_labels(SHELF_LABELS)["facings"]:
        label = facing.get("label")
        if label in (None, EXCLUDE):
            continue
        full = _evidence(lane, image, facing["box"])
        if full is None:
            continue
        shelf_full.append((label, full[0], full[1]))
        # A facing the live frame cuts to nothing is dropped rather than
        # scored at full resolution: the two columns have to be the same crops
        # at two sizes, or the resolution is not the only variable.
        small = _evidence(lane, live, [int(round(v * factor)) for v in facing["box"]])
        if small is not None:
            shelf_live.append((label, small[0], small[1]))

    engine = build_default_engine(EngineSettings(stock_dir=stock_dir))
    conf = LiveSettings().conf
    distractors: List[Evidence] = []
    photos = 0
    for directory in DISTRACTOR_DIRS:
        paths = sorted(
            path for path in glob.glob(os.path.join(directory, "*"))
            if os.path.splitext(path)[1].lower() in (".jpg", ".jpeg", ".png")
        )
        for path in paths:
            shelf_photo = cv2.imread(path)
            if shelf_photo is None:
                continue
            photos += 1
            frame = at_live_resolution(shelf_photo)
            for detection in engine.detect(frame, conf=conf):
                box = detection.get("box") or detection.get("bbox")
                if not box:
                    continue
                item = _evidence(lane, frame, [int(v) for v in box])
                if item is not None:
                    distractors.append(item)
    return {
        "version": CACHE_VERSION,
        "conf": conf,
        "photos": photos,
        "shelf_full": shelf_full,
        "shelf_live": shelf_live,
        "distractors": distractors,
    }


def _score(lane: LocalLaneAdapter, policy: DecisionPolicy, facings: Sequence[Facing]) -> Tuple[int, int]:
    """Shown and correct labels for one shelf column under one policy."""
    shown = correct = 0
    for label, ranked, read in facings:
        text = sorted(
            lane.text_lane.scores_from_text(read).items(), key=lambda kv: kv[1], reverse=True
        )
        answer = policy.decide(ranked, text, text_read=read)["best_sku"]
        if answer:
            shown += 1
            correct += answer == label
    return shown, correct


def _false_accepts(lane: LocalLaneAdapter, policy: DecisionPolicy,
                   distractors: Sequence[Evidence]) -> int:
    """Every accept on an unrelated shelf is a false accept by construction."""
    accepts = 0
    for ranked, read in distractors:
        text = sorted(
            lane.text_lane.scores_from_text(read).items(), key=lambda kv: kv[1], reverse=True
        )
        if policy.decide(ranked, text, text_read=read)["best_sku"]:
            accepts += 1
    return accepts


def main(argv: Optional[List[str]] = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--stock", default="stock-lopia-clean")
    parser.add_argument("--cache", default="", help="build once, reuse on later runs")
    parser.add_argument("--rebuild", action="store_true", help="ignore an existing cache")
    args = parser.parse_args(argv)

    cache: Optional[Dict[str, Any]] = None
    if args.cache and os.path.exists(args.cache) and not args.rebuild:
        with open(args.cache, "rb") as handle:
            cache = pickle.load(handle)
        if cache.get("version") != CACHE_VERSION:
            cache = None
    if cache is None:
        cache = build_cache(args.stock)
        if args.cache:
            os.makedirs(os.path.dirname(os.path.abspath(args.cache)), exist_ok=True)
            with open(args.cache, "wb") as handle:
                pickle.dump(cache, handle)

    lane = LocalLaneAdapter(args.stock)
    shelf_full: List[Facing] = cache["shelf_full"]
    shelf_live: List[Facing] = cache["shelf_live"]
    distractors: List[Evidence] = cache["distractors"]
    known = sum(1 for label, _, _ in shelf_full if label != UNKNOWN)

    print(f"catalog {args.stock}, {LIVE_LONG_SIDE}-long-side frames at detect_conf {cache['conf']}")
    print(f"image6: {len(shelf_full)} facings, {known} holding a catalog row, "
          f"{len(shelf_live)} of them cut from the live frame")
    print(f"distractors: {len(distractors)} crops from {cache['photos']} unrelated shelf photos")
    print()
    print(f"{'accept':>6} {'floor':>5} | {'shown':>5} {'corr':>4} {'prec':>6} {'cover':>6} "
          f"| {'shown':>5} {'corr':>4} {'prec':>6} | {'false':>5} {'rate':>6}")
    print(f"{'':>6} {'':>5} | {'--- full resolution ---':^25} "
          f"| {'---- live resolution ----':^25} | {'-- unknown shelves --':^14}")
    for accept in LOCAL_ACCEPTS:
        for floor in TEXT_MIN_SHAREDS:
            policy = DecisionPolicy(
                local_accept=accept, text_min_shared=floor,
                signatures=lane.text_lane.signatures,
            )
            full_shown, full_correct = _score(lane, policy, shelf_full)
            live_shown, live_correct = _score(lane, policy, shelf_live)
            false = _false_accepts(lane, policy, distractors)
            shown, correct = full_shown, full_correct
            rate = false / len(distractors) * 100
            shipped = "  <- shipped" if (accept, floor) == (LOCAL_ACCEPT, TEXT_MIN_SHARED) else ""
            print(
                f"{accept:>6.1f} {floor:>5} | {shown:>5} {correct:>4} "
                f"{correct / shown * 100 if shown else 0:>5.1f}% {correct / known * 100:>5.1f}% "
                f"| {live_shown:>5} {live_correct:>4} "
                f"{live_correct / live_shown * 100 if live_shown else 0:>5.1f}% "
                f"| {false:>5} {rate:>5.2f}%{shipped}"
            )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())