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

"""
Reconciliation (item vs price tag) and the first compliance rule family.

Two pure layers, no I/O:

- `reconcile()` compares one accepted identity with its tag and the catalog entry,
  and publishes a state plus the individual signals. It never changes the product
  identity: a disagreement is a second opinion, displayed next to the first.
- `evaluate_compliance()` turns settled evidence into findings. Every rule fails
  closed on evidence quality: a row whose tags are merely unreadable produces a
  capture hint, never a violation.
"""

from __future__ import annotations

import os
import re
import sys
import unicodedata
from dataclasses import dataclass, field
from typing import Any, Dict, Iterable, 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)

from tags import TAG_STATE_ABSENT, TAG_STATE_READABLE, TAG_STATE_UNREADABLE  # noqa: E402

RECONCILED_CONFIRMED = "confirmed"
RECONCILED_MISMATCH = "mismatch"
RECONCILED_UNVERIFIED = "unverified"
RECONCILED_NO_TAG = "no_tag"
RECONCILED_TAG_ONLY = "tag_only"

RULE_MISSING_TAG = "missing_price_tag"
RULE_MISSING_TAG_GROUP = "missing_tag_group"

# How far along a row a tag may name facings: enough to cover the product's own run
# of facings (measured: up to four facings, 70-120 px each), not enough to reach
# across the shelf. The cap exists because a product whose own tag was not detected
# would otherwise inherit whichever readable tag happens to be nearest.
TAG_REACH_FACINGS = 2.5
TAG_REACH_PX = 200
# The name similarity a tag needs before it may overrule the packaging channel.
TAG_OVERRIDE_SIMILARITY = 0.6
RULE_ORPHAN_TAG = "orphan_tag"
RULE_PRICE_MISMATCH = "price_mismatch"
RULE_DUPLICATE_TAG = "duplicate_tag"
RULE_UNREADABLE_TAGS = "unreadable_tags"

NAME_MATCH_THRESHOLD = 0.4
GROUP_MIN_FACINGS = 3
STABILITY_FRAMES = 3
STABILITY_WINDOW = 5


# ---------------------------------------------------------------------------
# text matching
# ---------------------------------------------------------------------------


def normalize_name(text: Optional[str]) -> str:
    if not text:
        return ""
    text = unicodedata.normalize("NFKC", text)
    text = re.sub(r"<[^>]*>", "", text)
    text = re.sub(r"[\s・、。,.\-–—_/]+", "", text)
    return text.lower()


def longest_common_substring_ratio(a: str, b: str) -> float:
    """Length of the longest common substring over the shorter string."""
    if not a or not b:
        return 0.0
    shorter, longer = (a, b) if len(a) <= len(b) else (b, a)
    best = 0
    for start in range(len(shorter)):
        for end in range(start + best + 1, len(shorter) + 1):
            if shorter[start:end] in longer:
                best = end - start
            else:
                break
    return best / len(shorter)


# ---------------------------------------------------------------------------
# reconciliation
# ---------------------------------------------------------------------------


@dataclass
class Signal:
    kind: str          # "barcode" | "name" | "price"
    agrees: bool
    item: Any
    tag: Any
    note: Optional[str] = None
    score: Optional[float] = None

    def as_dict(self) -> Dict[str, Any]:
        payload = {"kind": self.kind, "agrees": self.agrees, "item": self.item, "tag": self.tag}
        if self.note:
            payload["note"] = self.note
        if self.score is not None:
            payload["score"] = round(float(self.score), 3)
        return payload


def reconcile(
    identity: Optional[Dict[str, Any]],
    tag: Optional[Dict[str, Any]],
    product: Optional[Dict[str, Any]],
) -> Optional[Dict[str, Any]]:
    """Compare one facing with its tag.

    Returns None when there is nothing to reconcile (no accepted identity and no
    readable tag). A tag with no product above it is `tag_only`.
    """
    identity_state = (identity or {}).get("state")
    tag_state = (tag or {}).get("state", TAG_STATE_ABSENT)
    accepted = identity_state == "recognized"
    readable_tag = tag_state == TAG_STATE_READABLE

    if not accepted and not readable_tag:
        return None

    if accepted and tag_state == TAG_STATE_ABSENT:
        return {"state": RECONCILED_NO_TAG, "signals": [], "message": None}
    if accepted and tag_state == TAG_STATE_UNREADABLE:
        return {
            "state": RECONCILED_UNVERIFIED,
            "signals": [],
            "message": "tag found but its text was not read",
        }
    if not accepted and readable_tag:
        return {"state": RECONCILED_TAG_ONLY, "signals": [], "message": None}

    signals: List[Signal] = []

    # A barcode is the only exact signal; a checksummed code is not a guess. The
    # number a tag prints large beside the sale price is NOT a barcode - it is the
    # regular price (verified against the manufacturer: 4901577092381 for a tag
    # printing 357), so it is compared as a price below.
    item_barcode = str((product or {}).get("jan") or "").strip()
    tag_barcode = str((tag or {}).get("barcode") or "").strip()
    if item_barcode and tag_barcode:
        signals.append(Signal("barcode", item_barcode == tag_barcode, item_barcode, tag_barcode))

    item_name = normalize_name((product or {}).get("product_name"))
    tag_name = normalize_name((tag or {}).get("name"))
    if item_name and tag_name:
        ratio = longest_common_substring_ratio(item_name, tag_name)
        signals.append(
            Signal("name", ratio >= NAME_MATCH_THRESHOLD, item_name[:32], tag_name[:32],
                   note=f"match {ratio:.2f}", score=ratio)
        )

    item_price = (product or {}).get("price")
    tag_price = (tag or {}).get("price")
    tag_regular = (tag or {}).get("regular_price")
    if item_price is not None and (tag_price is not None or tag_regular is not None):
        candidates = [value for value in (tag_price, tag_regular) if value is not None]
        agrees = any(abs(float(item_price) - float(value)) < 0.5 for value in candidates)
        signals.append(Signal("price", agrees, item_price, tag_price if tag_price is not None else tag_regular))

    if not signals:
        return {
            "state": RECONCILED_UNVERIFIED,
            "signals": [],
            "message": "tag read, but nothing comparable in the catalog",
        }

    if any(signal.kind == "barcode" for signal in signals):
        decided = [signal for signal in signals if signal.kind == "barcode"][0]
        state = RECONCILED_CONFIRMED if decided.agrees else RECONCILED_MISMATCH
    elif any(signal.agrees for signal in signals):
        state = RECONCILED_CONFIRMED
    elif all(not signal.agrees for signal in signals):
        state = RECONCILED_MISMATCH
    else:
        state = RECONCILED_UNVERIFIED

    message = None
    if state == RECONCILED_MISMATCH:
        price_signal = next((signal for signal in signals if signal.kind == "price"), None)
        name_signal = next((signal for signal in signals if signal.kind == "name"), None)
        if price_signal is not None:
            message = f"tag says {price_signal.tag}, catalog says {price_signal.item}"
        elif name_signal is not None:
            message = "tag name does not match the catalog entry"
    return {"state": state, "signals": [signal.as_dict() for signal in signals], "message": message}


# ---------------------------------------------------------------------------
# compliance
# ---------------------------------------------------------------------------


@dataclass
class ComplianceSettings:
    group_min_facings: int = GROUP_MIN_FACINGS
    stability_frames: int = STABILITY_FRAMES
    stability_window: int = STABILITY_WINDOW
    # A row may only be judged when enough of its facings carry tag evidence
    # (readable or unreadable). Below this, the row is "not evaluated": claiming a
    # tag is missing because our detector found none would be a false violation.
    # 0.7 measured: at 0.5 a row where our own tag detection was patchy produced a
    # false "group without price tag" on a shelf where every facing has a tag.
    min_row_tag_coverage: float = 0.7


@dataclass
class Finding:
    rule: str
    severity: str                 # "violation" | "hint"
    row_index: Optional[int]
    box: List[int]
    count: int = 1
    skus: List[str] = field(default_factory=list)
    detail: Optional[str] = None

    def key(self) -> Tuple[Any, ...]:
        return (self.rule, self.row_index, self.box[0] // 32, self.box[1] // 32, self.count)

    def as_dict(self) -> Dict[str, Any]:
        payload = {
            "rule": self.rule,
            "severity": self.severity,
            "row_index": self.row_index,
            "box": list(self.box),
            "count": self.count,
            "skus": list(self.skus),
        }
        if self.detail:
            payload["detail"] = self.detail
        return payload


class ComplianceTracker:
    """Keeps findings only while they persist, so the UI never flickers."""

    def __init__(self, settings: Optional[ComplianceSettings] = None):
        self.settings = settings or ComplianceSettings()
        self._history: List[set] = []

    def update(self, findings: Sequence[Finding]) -> List[Finding]:
        current = {finding.key(): finding for finding in findings}
        self._history.append(set(current))
        if len(self._history) > self.settings.stability_window:
            self._history.pop(0)
        stable: List[Finding] = []
        for key, finding in current.items():
            seen = sum(1 for frame in self._history if key in frame)
            if seen >= self.settings.stability_frames:
                stable.append(finding)
        return stable

    def reset(self) -> None:
        self._history.clear()


def _sku_of_track(track: Dict[str, Any]) -> Optional[str]:
    identity = track.get("identity") or {}
    if identity.get("state") == "recognized" and identity.get("sku_id"):
        return str(identity["sku_id"])
    return None


def group_facings(
    row_tracks: Sequence[Dict[str, Any]],
    max_gap_ratio: float = 0.35,
    size_tolerance: float = 0.45,
) -> List[List[Dict[str, Any]]]:
    """Group a row's facings into products.

    A store prints one price tag per *product*, not per facing: measured on the
    demo shelf, a row carries 16-18 facings but only 4-5 tags, because the four
    facings of one mayonnaise share a single tag. Counting facings therefore
    invents missing tags on a shelf that is perfectly labelled, so the rules count
    *groups*.

    Identity wins over geometry when both facings were recognised: measured, a
    run of ピュアセレクト facings was split by geometry alone because one facing
    was detected narrower than its neighbours, and the product was then reported
    as missing the tag that was in fact sitting further along its own run.
    """
    ordered = sorted(
        (track for track in row_tracks if track.get("box") or track.get("raw_box")),
        key=lambda track: (track.get("box") or track.get("raw_box"))[0],
    )
    groups: List[List[Dict[str, Any]]] = []
    for track in ordered:
        box = track.get("box") or track.get("raw_box")
        if groups:
            previous = groups[-1][-1]
            pbox = previous.get("box") or previous.get("raw_box")
            gap = box[0] - pbox[2]
            width = max(1, pbox[2] - pbox[0])
            height = max(1, pbox[3] - pbox[1])
            width2 = max(1, box[2] - box[0])
            height2 = max(1, box[3] - box[1])
            sku, previous_sku = _sku_of_track(track), _sku_of_track(previous)
            close = gap <= max_gap_ratio * width
            if sku and previous_sku:
                same_product = sku == previous_sku and close
            else:
                same_product = close and (
                    abs(height2 - height) / height <= size_tolerance
                    and abs(width2 - width) / width <= size_tolerance
                )
            if same_product:
                groups[-1].append(track)
                continue
        groups.append([track])
    return groups


def cluster_into_rows(tracks: Sequence[Dict[str, Any]]) -> List[List[Dict[str, Any]]]:
    """Split facings into shelf rows by vertical overlap.

    `group_facings` is a *within-row* operation: given the whole frame it would
    merge the products stacked in one column across six shelves into a single
    "product". Identity assignment needs rows first.
    """
    rows: List[List[Dict[str, Any]]] = []
    for track in sorted(
        (track for track in tracks if track.get("box") or track.get("raw_box")),
        key=lambda track: (track.get("box") or track.get("raw_box"))[1],
    ):
        box = track.get("box") or track.get("raw_box")
        placed = False
        for row in rows:
            row_box = _group_box(row)
            overlap = min(row_box[3], box[3]) - max(row_box[1], box[1])
            if overlap > 0.5 * min(row_box[3] - row_box[1], box[3] - box[1]):
                row.append(track)
                placed = True
                break
        if not placed:
            rows.append([track])
    return rows


def _group_box(group: Sequence[Dict[str, Any]]) -> List[int]:
    boxes = [track.get("box") or track.get("raw_box") for track in group]
    return [
        min(box[0] for box in boxes),
        min(box[1] for box in boxes),
        max(box[2] for box in boxes),
        max(box[3] for box in boxes),
    ]


def evaluate_compliance(
    tracks: Sequence[Dict[str, Any]],
    tag_by_track: Dict[int, Dict[str, Any]],
    all_tags: Sequence[Dict[str, Any]],
    bands: Sequence[Any],
    reconciliations: Optional[Dict[int, Dict[str, Any]]] = None,
    settings: Optional[ComplianceSettings] = None,
) -> List[Finding]:
    """Findings from settled evidence. Rules with unreadable evidence stay silent.

    `tag_by_track` is the per-facing attachment; `all_tags` is every tag the pass
    found, so a tag nobody claimed can be reported as an orphan.
    """
    settings = settings or ComplianceSettings()
    findings: List[Finding] = []

    by_row: Dict[int, List[Dict[str, Any]]] = {}
    for track in tracks:
        row_index = _row_for(track.get("box") or track.get("raw_box"), bands)
        by_row.setdefault(row_index, []).append(track)

    for row_index, row_tracks in by_row.items():
        row_tracks.sort(key=lambda track: (track.get("box") or [0])[0])
        row_tags = [tag_by_track.get(int(track["track_id"]), {}) for track in row_tracks]
        readable = [tag for tag in row_tags if tag.get("state") == TAG_STATE_READABLE]
        evidenced = [
            tag for tag in row_tags if tag.get("state") in (TAG_STATE_READABLE, TAG_STATE_UNREADABLE)
        ]
        # Count products, not facings: one tag serves a run of identical facings.
        products = group_facings(row_tracks)
        product_tags = [
            [
                tag_by_track.get(int(track["track_id"]), {})
                for track in group
            ]
            for group in products
        ]
        products_evidenced = [
            group
            for group, tags in zip(products, product_tags)
            if any(tag.get("state") in (TAG_STATE_READABLE, TAG_STATE_UNREADABLE) for tag in tags)
        ]
        coverage = len(products_evidenced) / max(1, len(products))

        if not readable or coverage < settings.min_row_tag_coverage:
            # Fail closed on evidence quality. Either nothing in this row could be
            # read, or too few facings carry any tag evidence at all - in both cases
            # the honest answer is "not evaluated", never "tag missing".
            if evidenced:
                detail = (
                    "move closer: the price tags in this row could not be read"
                    if not readable
                    else (
                        f"tags detected for {len(products_evidenced)} of {len(products)} products "
                        f"in this row"
                    )
                )
                findings.append(
                    Finding(
                        rule=RULE_UNREADABLE_TAGS,
                        severity="hint",
                        row_index=row_index,
                        box=_row_box(row_tracks),
                        count=len(row_tracks),
                        detail=detail,
                    )
                )
            continue

        # missing_price_tag: one finding per row (the UI highlights the row), with
        # the products that lack a tag listed rather than one finding per facing.
        # A product's tag may sit anywhere along its run of facings, so a group
        # only counts as missing when no tag exists for its SKU elsewhere in the
        # row either. This is the domain rule, applied as a guard.
        skus_with_tags = {
            _sku_of_track(track)
            for track, tag in zip(row_tracks, row_tags)
            if tag.get("state") in (TAG_STATE_READABLE, TAG_STATE_UNREADABLE)
        } - {None}
        # A fragment of a product is not a product. Measured without identities: a
        # facing detected at 54 px wide beside 150 px neighbours was split off and
        # reported as missing the tag that belonged to its own run. Anything much
        # narrower than the row's typical product is not judged.
        widths = sorted(
            max(1, (_group_box(group)[2] - _group_box(group)[0])) for group in products
        )
        typical_width = widths[len(widths) // 2] if widths else 0
        missing_groups = [
            group
            for group, tags in zip(products, product_tags)
            if all(tag.get("state") == TAG_STATE_ABSENT for tag in tags)
            and not (_sku_of_track(group[0]) in skus_with_tags)
            and (_group_box(group)[2] - _group_box(group)[0]) >= 0.6 * typical_width
        ]
        if missing_groups:
            findings.append(
                Finding(
                    rule=RULE_MISSING_TAG,
                    severity="violation",
                    row_index=row_index,
                    box=_row_box([track for group in missing_groups for track in group]),
                    count=len(missing_groups),
                    skus=[_sku_of(group[0]) for group in missing_groups],
                    detail=(
                        f"{len(missing_groups)} of {len(products)} products in this row "
                        f"have no price tag"
                    ),
                )
            )

        # missing_tag_group: consecutive products without a tag
        missing_ids = {id(group) for group in missing_groups}
        run: List[List[Dict[str, Any]]] = []
        for group, tags in zip(products, product_tags):
            if id(group) in missing_ids:
                run.append(group)
                continue
            if len(run) >= settings.group_min_facings:
                findings.append(
                    _group_finding(row_index, [track for g in run for track in g], len(run))
                )
            run = []
        if len(run) >= settings.group_min_facings:
            findings.append(_group_finding(row_index, [track for g in run for track in g], len(run)))

    # price_mismatch from reconciliation
    for track in tracks:
        track_id = int(track["track_id"])
        reconciliation = (reconciliations or {}).get(track_id)
        if not reconciliation or reconciliation.get("state") != RECONCILED_MISMATCH:
            continue
        signals = reconciliation.get("signals", [])
        if not any(signal.get("kind") == "price" for signal in signals):
            continue
        # A mismatch needs one trustworthy anchor before it is a finding. Measured:
        # a wrong identity plus a garbled tag read (name match 0.11, no JAN)
        # produced a confident-looking "tag says 90, catalog says 188" that was
        # really just noise. With an anchor - the tag's JAN, or a name that
        # actually resembles the product - the disagreement is worth reporting.
        tag_for_track = tag_by_track.get(track_id, {})
        name_signal = next(
            (signal for signal in signals if signal.get("kind") == "name"), None
        )
        # The anchor is the tag's own arithmetic agreeing, or a name that actually
        # resembles the product. (It used to be `tag.jan`, which held a *price*.)
        flags = set(tag_for_track.get("flags") or [])
        corroborated = bool({"tax_consistent", "discount_consistent"} & flags)
        name_score = float(name_signal.get("score") or 0.0) if name_signal else 0.0
        has_anchor = corroborated or name_score >= 0.35
        if not has_anchor:
            findings.append(
                Finding(
                    rule=RULE_UNREADABLE_TAGS,
                    severity="hint",
                    row_index=_row_for(track.get("box") or track.get("raw_box"), bands),
                    box=list(track.get("box") or track.get("raw_box") or [0, 0, 0, 0]),
                    count=1,
                    skus=[_sku_of(track)],
                    detail="the tag under this product could not be read well enough to compare",
                )
            )
            continue
        findings.append(
            Finding(
                rule=RULE_PRICE_MISMATCH,
                severity="violation",
                row_index=_row_for(track.get("box") or track.get("raw_box"), bands),
                box=list(track.get("box") or track.get("raw_box") or [0, 0, 0, 0]),
                count=1,
                skus=[_sku_of(track)],
                detail=reconciliation.get("message"),
            )
        )

    # orphan_tag: a readable tag that no facing claims (a tag with no product above
    # it is either an empty slot with a leftover tag or a mis-clustered read).
    attached_ids = {
        tag.get("id") for tag in tag_by_track.values() if tag.get("id")
    }
    for tag in all_tags:
        if tag.get("state") != TAG_STATE_READABLE or not tag.get("id"):
            continue
        if tag["id"] in attached_ids:
            continue
        findings.append(
            Finding(
                rule=RULE_ORPHAN_TAG,
                severity="violation",
                row_index=None,
                box=list(tag.get("box") or [0, 0, 0, 0]),
                count=1,
                detail="price tag with no product above it",
            )
        )

    return findings


def _group_finding(row_index: int, run: Sequence[Dict[str, Any]], products: int = 0) -> Finding:
    boxes = [track.get("box") or track.get("raw_box") or [0, 0, 0, 0] for track in run]
    x1 = min(box[0] for box in boxes)
    y1 = min(box[1] for box in boxes)
    x2 = max(box[2] for box in boxes)
    y2 = max(box[3] for box in boxes)
    return Finding(
        rule=RULE_MISSING_TAG_GROUP,
        severity="violation",
        row_index=row_index,
        box=[x1, y1, x2, y2],
        count=len(run),
        skus=[_sku_of(track) for track in run],
        detail=(
            f"{products or len(run)} products in a row without a price tag"
        ),
    )


def _row_box(tracks: Sequence[Dict[str, Any]]) -> List[int]:
    boxes = [track.get("box") or track.get("raw_box") or [0, 0, 0, 0] for track in tracks]
    return [
        min(box[0] for box in boxes),
        min(box[1] for box in boxes),
        max(box[2] for box in boxes),
        max(box[3] for box in boxes),
    ]


def _row_for(box: Optional[Sequence[int]], bands: Sequence[Any]) -> int:
    """Which rail band (i.e. row) a facing sits above."""
    if not box or not bands:
        return 0
    x_center = (box[0] + box[2]) / 2.0
    y_bottom = box[3]
    best_index = 0
    best_distance = None
    for band in bands:
        bx1, by1, bx2, by2 = band.box
        if not (bx1 - 40 <= x_center <= bx2 + 40):
            continue
        distance = abs(by1 - y_bottom)
        if best_distance is None or distance < best_distance:
            best_distance = distance
            best_index = band.row_index
    return best_index


def _sku_of(track: Dict[str, Any]) -> str:
    identity = track.get("identity") or {}
    return str(identity.get("sku_id") or "")


def best_catalog_match(
    name: Optional[str],
    catalog_names: Dict[str, str],
    min_similarity: float = 0.45,
) -> Tuple[Optional[str], float]:
    """The catalog entry a printed tag name most resembles."""
    if not name:
        return None, 0.0
    normalized = normalize_name(name)
    best_sku, best_score = None, 0.0
    for sku_id, catalog_name in catalog_names.items():
        score = longest_common_substring_ratio(normalized, catalog_name)
        if score > best_score:
            best_sku, best_score = sku_id, score
    if best_score < min_similarity:
        return None, best_score
    return best_sku, best_score


def identities_from_tags(
    tracks: Sequence[Dict[str, Any]],
    tag_by_track: Dict[int, Dict[str, Any]],
    catalog_by_sku: Dict[str, Dict[str, Any]],
    min_similarity: float = 0.40,
    all_tags: Optional[Sequence[Dict[str, Any]]] = None,
) -> Dict[int, Dict[str, Any]]:
    """Identify a product from the tag printed under it.

    The shelf-edge tag carries the store's own name for the product in clean
    printed text, and one tag serves a whole run of facings. That makes it the
    most reliable identity source on the shelf - far more so than recognising the
    packaging, which is fine-grained (twenty near-identical cartons), occluded,
    and distorted by glare and angle.

    Measured with an off-the-shelf DINOv2 embedding: identical products do cluster
    (two orange juices, two apple juices, two almond drinks) but fine-grained
    cases do not - 32 cartons collapsed into one cluster and nine different
    bottled drinks grouped by *container shape*.

    Assignment is by *tag position*, not by geometric grouping: the tags on a rail
    mark where each product sits, so a facing belongs to the nearest tag along the
    row. Grouping facings by size and spacing was measured to merge different
    products (a 945 px "product" spanning half a shelf), which then swallowed the
    wrong tag.
    """
    catalog_names = {
        sku_id: normalize_name(product.get("product_name"))
        for sku_id, product in catalog_by_sku.items()
        if product.get("product_name")
    }
    if not catalog_names:
        return {}

    tags = [dict(tag) for tag in (all_tags or []) if tag.get("box")]
    for tag in tags:
        if tag.get("sku_from_name") is None:
            sku_id, score = best_catalog_match(tag.get("name"), catalog_names, min_similarity)
            tag["sku_from_name"] = sku_id
            tag["sku_score"] = score
    if not tags:
        return {}

    found: Dict[int, Dict[str, Any]] = {}
    for row in cluster_into_rows(tracks):
        row_box = _group_box(row)
        row_height = max(1, row_box[3] - row_box[1])
        # The rail's tags sit just under the row they describe.
        row_tags = [
            tag
            for tag in tags
            if row_box[3] - 0.5 * (tag["box"][3] - tag["box"][1])
            <= tag["box"][1]
            <= row_box[3] + 1.5 * (tag["box"][3] - tag["box"][1])
        ]
        if not row_tags:
            continue
        for track in row:
            box = track.get("box") or track.get("raw_box")
            center = (box[0] + box[2]) / 2.0
            facing_width = max(1, box[2] - box[0])
            # A tag names the product at its own position on the rail. Without a
            # distance cap a facing whose own tag was not detected inherits the
            # nearest readable tag's product, which can belong to a different
            # product several columns away.
            limit = max(TAG_REACH_FACINGS * facing_width, TAG_REACH_PX)
            candidates = [
                tag for tag in row_tags
                if abs(center - (tag["box"][0] + tag["box"][2]) / 2.0) <= limit
            ]
            if not candidates:
                continue
            nearest = min(
                candidates,
                key=lambda tag: abs(center - (tag["box"][0] + tag["box"][2]) / 2.0),
            )
            sku_id = nearest.get("sku_from_name")
            if not sku_id:
                continue
            found[int(track["track_id"])] = {
                "sku_id": sku_id,
                "score": round(float(nearest.get("sku_score") or 0.0), 3),
                "source": "tag_name",
            }
    return found


def attach_tags_to_boxes(
    tag_service: Any,
    image: Any,
    boxes: Sequence[Dict[str, Any]],
    catalog_by_sku: Dict[str, Dict[str, Any]],
    identities: Optional[Dict[int, Dict[str, Any]]] = None,
) -> Dict[str, Any]:
    """One-shot tag pass for a still image (photo mode / `?tags=true`).

    Runs the passes synchronously - a still is not a stream, so there is no frame
    loop to protect - then attaches tags, reconciles, and evaluates compliance.
    """
    image_shape = image.shape
    # A still has no later frame to pick up the remaining bands, and `plan` only
    # hands out a few per call (a stream must stay responsive). Drain them, with a
    # cap so a huge shelf cannot turn one request into minutes of OCR. Draining
    # also feeds the vote: a band read in two rounds is two observations.
    bands: List[Any] = []
    rounds = 0
    while rounds < 4:
        planned = tag_service.plan(boxes, image_shape)
        if not planned:
            break
        for band in planned:
            # Two independent reads (different upscale) so the vote has evidence
            # to work with; a single read has nothing to vote against.
            tag_service.run_pass(image, band, scale_bias=1.0)
            tag_service.run_pass(image, band, scale_bias=1.7)
            bands.append(band)
        rounds += 1

    tracks = [
        {
            "track_id": int(box["id"]),
            "box": list(box["box"]),
            "identity": (identities or {}).get(int(box["id"])) or {},
            "tag_identity": {},
        }
        for box in boxes
    ]
    tag_by_track, all_tags, counts = tag_service.attach(tracks)

    # Tag-first identity, applied before reconciliation so the comparison below
    # sees the identity the tag itself provides.
    tag_identities = identities_from_tags(
        tracks, tag_by_track, catalog_by_sku, all_tags=all_tags
    )
    for track in tracks:
        derived = tag_identities.get(int(track["track_id"]))
        if not derived:
            continue
        # Always keep the tag's claim, even when packaging recognition already
        # named the product: two independent identity channels that can be
        # compared are worth more than one that silently wins. A disagreement is
        # itself a finding (a misplaced tag, or a product in the wrong place).
        track["tag_identity"] = dict(derived)
        current = track["identity"] or {}
        if current.get("state") == "recognized":
            if current.get("sku_id") != derived["sku_id"]:
                # Only a corroborated tag, or a name that closely resembles the
                # product, may overrule the packaging channel. A weak 0.40 name
                # match used to be enough, which is not evidence.
                tag_flags = set((tag_by_track.get(int(track["track_id"])) or {}).get("flags") or [])
                strong = bool({"tax_consistent", "discount_consistent"} & tag_flags) or \
                    float(derived.get("score") or 0.0) >= TAG_OVERRIDE_SIMILARITY
                if not strong:
                    track["identity_conflict"] = {
                        "packaging": current.get("sku_id"),
                        "tag": derived["sku_id"],
                        "packaging_score": current.get("score"),
                        "note": "tag name match too weak to overrule packaging",
                    }
                    continue
                # Two channels disagree. Measured on the demo shelf: where both
                # named the same facing they agreed only 69% of the time, and the
                # packaging side of every disagreement carried a low score - so
                # the store's own label wins, and the disagreement is recorded
                # rather than hidden (a misplaced tag is a real finding).
                track["identity_conflict"] = {
                    "packaging": current.get("sku_id"),
                    "tag": derived["sku_id"],
                    "packaging_score": current.get("score"),
                }
                track["identity"] = {
                    "state": "recognized",
                    "sku_id": derived["sku_id"],
                    "identity_level": "tag_over_packaging",
                    "confidence": derived["score"],
                    "score": derived["score"],
                    "source": "tag_name",
                    "candidates": [],
                    "attempts": 1,
                    "method": "tag_name",
                }
            continue
        track["identity"] = {
            "state": "recognized",
            "sku_id": derived["sku_id"],
            "identity_level": "tag_derived",
            "confidence": derived["score"],
            "score": derived["score"],
            "source": "tag_name",
            "candidates": [],
            "attempts": 1,
            "method": "tag_name",
        }

    reconciliations: Dict[int, Dict[str, Any]] = {}
    for track in tracks:
        track_id = track["track_id"]
        tag = tag_by_track.get(track_id)
        product = catalog_by_sku.get((track["identity"] or {}).get("sku_id") or "")
        decision = reconcile(track["identity"], tag, product)
        if decision is not None:
            reconciliations[track_id] = decision

    findings = evaluate_compliance(tracks, tag_by_track, all_tags, bands, reconciliations)
    summary = summarize_findings(findings)
    return {
        "by_box": tag_by_track,
        "tags": all_tags,
        "counts": counts,
        "reconciliations": reconciliations,
        "tag_identities": tag_identities,
        "bands": [{"row_index": band.row_index, "box": band.box} for band in bands],
        "bands_objects": list(bands),
        "compliance": {
            "violations": summary["violations"],
            "by_rule": summary["by_rule"],
            "rows": [
                {
                    "row_index": finding.row_index,
                    "rule": finding.rule,
                    "box": finding.box,
                    "missing": finding.count,
                    "skus": finding.skus,
                }
                for finding in findings
                if finding.severity == "violation"
            ],
            "hints": summary["hints"],
        },
    }


def reconcile_bundle(
    bundle: Dict[str, Any],
    detections: Sequence[Dict[str, Any]],
    catalog_by_sku: Dict[str, Dict[str, Any]],
) -> Dict[str, Any]:
    """Re-run reconciliation and compliance against a cached tag bundle.

    Used when identities arrive after the tag pass (photo mode: recognize paints
    boxes, classify names them). Without this, every tag would look like `tag_only`
    because it was compared against an identity that did not exist yet.
    """
    tracks = [
        {
            "track_id": int(detection["id"]),
            "box": list(detection["box"]),
            "identity": detection.get("identity") or {},
        }
        for detection in detections
    ]
    tag_by_track = bundle.get("by_box") or {}
    reconciliations: Dict[int, Dict[str, Any]] = {}
    for track in tracks:
        track_id = track["track_id"]
        tag = tag_by_track.get(track_id)
        product = catalog_by_sku.get((track["identity"] or {}).get("sku_id") or "")
        decision = reconcile(track["identity"], tag, product)
        if decision is not None:
            reconciliations[track_id] = decision

    findings = evaluate_compliance(
        tracks, tag_by_track, bundle.get("tags") or [], bundle.get("bands_objects") or [], reconciliations
    )
    summary = summarize_findings(findings)
    return {
        "reconciliations": reconciliations,
        "compliance": {
            "violations": summary["violations"],
            "by_rule": summary["by_rule"],
            "rows": [
                {
                    "row_index": finding.row_index,
                    "rule": finding.rule,
                    "box": finding.box,
                    "missing": finding.count,
                    "skus": finding.skus,
                }
                for finding in findings
                if finding.severity == "violation"
            ],
            "hints": summary["hints"],
        },
    }


def summarize_findings(findings: Sequence[Finding]) -> Dict[str, Any]:
    by_rule: Dict[str, int] = {}
    for finding in findings:
        if finding.severity != "violation":
            continue
        by_rule[finding.rule] = by_rule.get(finding.rule, 0) + 1
    return {
        "violations": sum(by_rule.values()),
        "by_rule": by_rule,
        "hints": [finding.as_dict() for finding in findings if finding.severity != "violation"],
    }