product-recognition-demo/recognization/docs/research/2026-09-29-shelf-intelligence/far_unknown.py

"""The four named policies on the far-unknown crops, priced exactly.

`far-unknown-lines.json` sweeps the two open-set gates as a grid - LOCAL_ACCEPT x
TEXT_MIN_SHARED, one shipped `DecisionPolicy` per cell, no fallback. That table
says where the gates cross; it cannot say what a *named* policy costs on a shelf
holding nothing in the catalog. Near-argmax is not a threshold at all - it keeps a
geometric candidate the gate refused, behind two refusals of its own - and no cell
in that grid expresses it.

So the four policies go through `analyze.decide`, the same call the frozen
observation table uses, over the same cached crops `sweep_accept_lines.py` built.
`gate_accepts` is the shipped `DecisionPolicy` on identical evidence, so the gap
between the two counts per policy is exactly what the fallback adds. Every accept
on an unrelated Wikimedia shelf is a false accept by construction.

Run from the recognition root:

    .venv/bin/python docs/research/2026-09-29-shelf-intelligence/far_unknown.py
"""
from __future__ import annotations

import argparse
import hashlib
import json
import pickle
import sys
from pathlib import Path

DIR = Path(__file__).resolve().parent
ROOT = Path(__file__).resolve().parents[3]

sys.path.insert(0, str(ROOT / "code"))
sys.path.insert(0, str(DIR))

import analyze  # noqa: E402
from eval_identity import _catalog_rows, catalog_hash, code_revision  # noqa: E402
from identity_lanes import DecisionPolicy, TextLane, _ranked  # noqa: E402
from sweep_accept_lines import CACHE_VERSION  # noqa: E402

CACHE = "data/evaluations/2026-09-29-session/scratch/accept_line_cache.pkl"
LINES = "evidence/far-unknown-lines.json"
OBSERVATIONS = "evidence/observations.json"


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--stock", default="stock-lopia-clean")
    parser.add_argument("--cache", default=CACHE)
    parser.add_argument("--out", default=str(DIR / "evidence/far-unknown-policies.json"))
    args = parser.parse_args(argv)

    # Hash the exact bytes that get unpickled, so the recorded digest and the
    # measured crops cannot come from two different reads of a rewritten file.
    blob = Path(args.cache).read_bytes()
    cache = pickle.loads(blob)
    if cache.get("version") != CACHE_VERSION:
        raise SystemExit(
            f"cache version {cache.get('version')!r} != sweep's {CACHE_VERSION}; "
            "rebuild it with code/sweep_accept_lines.py --rebuild"
        )

    # No OCR engine: the reads are already in the cache, so `read()` is never
    # called and building a recognizer would only add a dependency to fail.
    lane = TextLane.from_catalog(_catalog_rows(args.stock), args.stock)
    signatures = lane.signatures

    distractors = cache["distractors"]
    observations = []
    for ranked, read in distractors:
        text = lane.scores_from_text(read)
        # The cached rank is passed through untouched, and the dict scores are
        # ordered exactly as the grid sweep ordered them, so the four policies
        # here and the four columns there read identical evidence.
        assert _ranked(text) == sorted(text.items(), key=lambda kv: kv[1], reverse=True)
        observations.append({"rank": ranked, "text": text, "read": read})
    assert len(observations) == len(distractors)

    results = {}
    for policy, threshold in analyze.POLICIES.items():
        gate = DecisionPolicy(local_accept=threshold, signatures=signatures)
        accepts = fallback_only = with_rank = 0
        by_source = {"local": 0, "text": 0}
        for o in observations:
            if o["rank"]:
                with_rank += 1
            decision = gate.decide(o["rank"], o["text"], text_read=o["read"])
            answer = analyze.decide(o, policy, signatures)
            if not answer:
                continue
            accepts += 1
            if decision["best_sku"]:
                # The fallback only runs when the gate refused, so a gated
                # answer must survive `analyze.decide` unchanged.
                assert answer == decision["best_sku"]
                by_source[decision["source"]] += 1
            else:
                fallback_only += 1
        # Accounting: every crop decided once, accepts split exhaustively, and
        # the fallback belongs to near-argmax alone. A weaker floor accepts
        # weakly more, so the counts rise as the threshold falls.
        assert accepts - fallback_only >= 0
        assert by_source["local"] + by_source["text"] + fallback_only == accepts
        assert fallback_only == 0 or policy == "near-argmax"
        results[policy] = {
            "local_accept": threshold,
            "fallback": policy == "near-argmax",
            "accepts": accepts,
            "gate_accepts": accepts - fallback_only,
            "fallback_accepts": fallback_only,
            "by_source": by_source,
            "rate": accepts / len(observations),
            "rate_percent": round(accepts / len(observations) * 100, 4),
            "crops_with_a_local_candidate": with_rank,
        }

    ordered = [results[name]["accepts"] for name in analyze.POLICIES]
    assert ordered == sorted(ordered), f"accepts not monotone in the accept floor: {ordered}"

    crops = len(observations)
    report = {
        "question": (
            "False accepts per named analyze policy on unrelated shelf crops, "
            "analyze.decide on the sweep_accept_lines cache."
        ),
        "method": {
            "decision": "analyze.decide(o, policy, signatures) - the frozen analysis's own call",
            "policies": {name: analyze.POLICIES[name] for name in analyze.POLICIES},
            "gate_for_comparison": "identity_lanes.DecisionPolicy(local_accept=threshold, signatures)",
            "text_scores": "TextLane.scores_from_text on the cached OCR read",
            "fallback": (
                "analyze.decide's near-argmax extra fallback only: keep rank[0] when the gate "
                "refused, unless the read contradicts that row's size or the text prefers another row"
            ),
        },
        "provenance": {
            "cache": {
                "path": args.cache,
                "sha256": hashlib.sha256(blob).hexdigest(),
                "bytes": len(blob),
                "version": cache["version"],
                "detect_conf": cache["conf"],
                "records": ["version", "conf", "photos", "shelf_full", "shelf_live", "distractors"],
                "not_recorded": [
                    "stock_dir",
                    "catalog_hash",
                    "code_revision",
                    "source image paths or hashes",
                    "per-photo crop counts",
                ],
                "note": (
                    "The cache attests no catalog, so the catalog match below is read off the "
                    "current worktree, not out of the cache. The read is unpickled, not rebuilt."
                ),
            },
            "catalog": {
                "dir": args.stock,
                "catalog_hash": catalog_hash(args.stock),
            },
            "code_revision": code_revision(),
            "signatures": {
                "rows": len(signatures),
                "sha256": hashlib.sha256(
                    json.dumps(signatures, sort_keys=True, ensure_ascii=False).encode()
                ).hexdigest(),
            },
        },
        "counts": {
            "photos": cache["photos"],
            "crops": crops,
            "crops_with_a_local_candidate": sum(1 for o in observations if o["rank"]),
            "crops_with_no_local_candidate": sum(1 for o in observations if not o["rank"]),
            "crops_with_a_raw_read": sum(1 for o in observations if o["read"]),
            # A read can be non-empty and still score nothing: `scores_from_text`
            # returns {} under four normalized characters.
            "crops_scored_by_text": sum(1 for o in observations if o["text"]),
        },
        "policies": results,
    }

    # Context, not an accounting input: the older grid table's same-threshold
    # cells, whose producer is not in the tree, so its settings stay unverified.
    lines_path = DIR / LINES
    lines = json.loads(lines_path.read_text()) if lines_path.is_file() else {}
    report["cross_check_far_unknown_lines"] = {
        "source": LINES,
        "note": "accept-threshold cells only, no fallback; producer script not in the tree",
        "cells": {
            name: {
                "lines_accepts": line["accepts"],
                "lines_crops": line["crops"],
                "here_gate_accepts": results[name]["gate_accepts"],
                "agrees": line["accepts"] == results[name]["gate_accepts"]
                and line["crops"] == crops,
            }
            for name, row in results.items()
            if (line := lines.get(str(row["local_accept"])))
        },
    }

    # What the cache cannot attest, read off the frozen table instead: the lane
    # built here carries the same 12 names the observation table was scored on.
    frozen = json.loads((DIR / OBSERVATIONS).read_text())
    report["provenance"]["catalog"].update({
        "matches_observations": frozen["signatures"] == signatures,
        "observations_catalog_hash": frozen["provenance"]["catalog"],
        "observations_code_revision": frozen["provenance"]["code"],
    })
    report["limitations"] = [
        "The cache records no catalog, code revision or image list, so 'same catalog as the "
        "frozen observations' is read off the current worktree, not attested by the cache.",
        f"Only {report['counts']['crops_with_a_local_candidate']} of {crops} crops produced a local "
        f"candidate; the other {report['counts']['crops_with_no_local_candidate']} are decided by the "
        "text lane alone or refused before any gate applies, so the accept floor is exercised on a "
        "small minority of the rows and the rates are dominated by empty evidence.",
        "The catalog is 12 rows from one store. A false-accept rate is a function of catalog size, so "
        "these numbers do not transfer to a larger or differently-stocked catalog.",
        "Crops come from one detection pass at detect_conf "
        f"{cache['conf']} on 1280-long-side frames, so the crop population is set by the detector, "
        "not only by the identity gate.",
        "The cache has no per-photo breakdown, so no photo-level rate and no check that one image is "
        "not carrying the whole count.",
        "The near-argmax fallback is a research construct in analyze.py, not shipped code; its row "
        "prices a policy that does not exist on the server yet.",
        "far-unknown-lines.json's producer is not in the tree, so agreement with it is corroboration, "
        "not verification.",
    ]

    Path(args.out).write_text(json.dumps(report, indent=1, ensure_ascii=False) + "\n")

    print(f"cache {args.cache} sha256 {report['provenance']['cache']['sha256'][:16]}… "
          f"v{cache['version']} conf {cache['conf']}")
    print(f"catalog {args.stock}: {len(signatures)} rows, hash "
          f"{report['provenance']['catalog']['catalog_hash'][:16]}…, matches observations "
          f"{report['provenance']['catalog']['matches_observations']}")
    print(f"{cache['photos']} unrelated shelf photos, {crops} crops, "
          f"{report['counts']['crops_with_a_local_candidate']} with a local candidate")
    print()
    print(f"{'policy':>15} {'accept':>7} {'fallb':>6} {'accepts':>8} {'gate':>6} "
          f"{'local':>6} {'text':>6} {'rate':>8}")
    for name, row in results.items():
        print(f"{name:>15} {row['local_accept']:>7.1f} {str(row['fallback']):>6} "
              f"{row['accepts']:>8} {row['gate_accepts']:>6} {row['by_source']['local']:>6} "
              f"{row['by_source']['text']:>6} {row['rate_percent']:>7.4f}%")
    print()
    print(f"wrote {args.out}")
    return 0


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