feat(ocr): add vnext SidecarPass

paperforge/worker/ocr_figure_vnext_sidecar_pass.py

@@ -0,0 +1,124 @@
+from __future__ import annotations
+
+from .ocr_figure_vnext_types import ClaimProposal, PassReport, ResourceRef
+
+
+def _resource_page(block: dict) -> int | None:
+    """Extract page number from a block dict (page or page_num)."""
+    page = block.get("page")
+    if page is None:
+        page = block.get("page_num")
+    return int(page) if page is not None else None
+
+
+class SidecarPass:
+    """Fallback pass matching narrow captions to same-page assets.
+
+    The PrimarySamePagePass targets wide legends with high-confidence
+    scoring.  Narrow captions (aligned in a side column, width <60 % of
+    page width) often score poorly because they lack the spatial
+    signatures of a full-width figure caption.  SidecarPass rescues
+    these by partitioning same-page assets into caption-aligned bands
+    and claiming each band's assets for its narrow caption.
+    """
+
+    name = "sidecar"
+
+    def run(self, state):
+        report = PassReport(pass_name=self.name)
+        from . import ocr_figures
+
+        sidecar_candidates = getattr(state.candidate_index, "sidecar_candidates", None) or {}
+
+        for page, narrow_captions in sidecar_candidates.items():
+            page_assets = [a for a in state.corpus.raw_assets if _resource_page(a) == page]
+            if not page_assets:
+                continue
+
+            # Compute page_height from blocks on this page (same pattern as
+            # ocr_figures.py:_sidecar_page_promotion, line 4117).
+            page_blocks = [b for b in state.corpus.blocks if _resource_page(b) == page]
+            page_height = float(
+                max((b.get("bbox") or [0, 0, 0, 0])[3] for b in page_blocks) or 1200
+            )
+
+            bands = ocr_figures._partition_assets_by_caption_bands(
+                narrow_captions, page_assets, page_height
+            )
+
+            for caption in narrow_captions:
+                cid = str(caption.get("block_id", ""))
+
+                # Skip captions already protected by an earlier pass
+                protected = False
+                for match in state.matches:
+                    if str(match.get("legend_block_id", "")) == cid:
+                        if ocr_figures._has_protected_figure_ownership(match):
+                            protected = True
+                            break
+                if protected:
+                    continue
+
+                band_assets = bands.get(cid, [])
+                if not band_assets:
+                    continue
+
+                asset_refs = []
+                for asset in band_assets:
+                    apage = _resource_page(asset)
+                    if apage is None:
+                        continue
+                    asset_refs.append(ResourceRef(
+                        kind="asset", page=apage, block_id=str(asset.get("block_id", "")),
+                    ))
+
+                if not asset_refs:
+                    continue
+
+                legend_ref = ResourceRef(kind="legend", page=page, block_id=cid)
+                figure_text = str(caption.get("text", ""))
+                figure_no = ocr_figures._extract_figure_number(figure_text)
+
+                proposal = ClaimProposal(
+                    pass_name=self.name,
+                    figure_no=figure_no,
+                    claim_type="match",
+                    legends=[legend_ref],
+                    assets=asset_refs,
+                    groups=[],
+                    confidence=0.3,
+                    evidence_rank=5,
+                    reason="sidecar_match",
+                    diagnostics={"caption_block_id": cid, "page": page},
+                )
+
+                conflict = state.ledger.try_claim_assets(
+                    proposal.assets, owner=proposal.legends[0], reason=proposal.reason,
+                )
+                if conflict is not None:
+                    report.conflicts.append(conflict)
+                    report.rejected.append(proposal)
+                    continue
+
+                namespace = ocr_figures._extract_figure_namespace(figure_text)
+                if figure_no is None:
+                    figure_id = f"figure_unknown_{len(state.matches):03d}"
+                else:
+                    figure_id = ocr_figures._format_figure_id(namespace, figure_no)
+
+                match_record = {
+                    "legend_block_id": cid,
+                    "figure_id": figure_id,
+                    "figure_number": figure_no,
+                    "settlement_type": "sidecar",
+                    "flags": ["sidecar_match"],
+                    "page": page,
+                    "matched_assets": [ocr_figures._project_asset_record(a) for a in band_assets],
+                    "evidence": [],
+                    "pass_name": self.name,
+                }
+
+                state.accept_match(proposal, match_record)
+                report.accepted.append(proposal)
+
+        return report

tests/test_ocr_figure_vnext_sidecar_pass.py

@@ -0,0 +1,119 @@
+from __future__ import annotations
+
+from paperforge.worker.ocr_figure_vnext_corpus import FigureCandidateIndex, FigureCorpus
+from paperforge.worker.ocr_figure_vnext_passes import PrimarySamePagePass
+from paperforge.worker.ocr_figure_vnext_sidecar_pass import SidecarPass
+from paperforge.worker.ocr_figure_vnext_state import FigurePipelineState, OwnershipLedger
+from paperforge.worker.ocr_figure_vnext_types import ResourceRef
+
+
+def test_sidecar_pass_matches_narrow_captions_to_asset_bands(monkeypatch):
+    """Three narrow captions on page 5 with three assets at different y-bands.
+
+    PrimarySamePagePass misses (monkeypatched scoring), SidecarPass rescues
+    each narrow caption into a sidecar match.
+    """
+    blocks = [
+        {"block_id": "c1", "page": 5, "role": "figure_caption",
+         "text": "Figure 1. First caption", "bbox": [100, 100, 300, 130]},
+        {"block_id": "c2", "page": 5, "role": "figure_caption",
+         "text": "Figure 2. Second caption", "bbox": [100, 300, 310, 330]},
+        {"block_id": "c3", "page": 5, "role": "figure_caption",
+         "text": "Figure 3. Third caption", "bbox": [100, 500, 300, 530]},
+        {"block_id": "a1", "page": 5, "role": "figure_asset",
+         "bbox": [50, 0, 250, 90]},
+        {"block_id": "a2", "page": 5, "role": "figure_asset",
+         "bbox": [50, 180, 260, 270]},
+        {"block_id": "a3", "page": 5, "role": "figure_asset",
+         "bbox": [60, 400, 270, 480]},
+    ]
+    corpus = FigureCorpus.from_blocks(blocks, page_width=1200)
+    index = FigureCandidateIndex.from_corpus(corpus)
+
+    assert 5 in index.sidecar_candidates
+    assert len(index.sidecar_candidates[5]) == 3
+
+    state = FigurePipelineState(corpus=corpus, candidate_index=index, ledger=OwnershipLedger())
+
+    # Ensure PrimarySamePagePass misses — narrow captions score poorly for same-page
+    monkeypatch.setattr(
+        "paperforge.worker.ocr_figures._score_legend_to_group",
+        lambda *a, **k: {"score": 0.0, "decision": "missed", "evidence": []},
+    )
+    PrimarySamePagePass().run(state)
+
+    report = SidecarPass().run(state)
+
+    assert len(report.accepted) >= 1
+    sidecar_matches = [m for m in state.matches if m.get("settlement_type") == "sidecar"]
+    assert len(sidecar_matches) >= 1
+
+    matched_ids = {m.get("legend_block_id") for m in sidecar_matches}
+    assert matched_ids == {"c1", "c2", "c3"}
+
+    for match in sidecar_matches:
+        assert len(match.get("matched_assets", [])) >= 1
+        assert "sidecar_match" in match.get("flags", [])
+
+
+def test_sidecar_pass_skips_pages_with_fewer_than_two_narrow_captions():
+    """Single narrow caption on a page — no sidecar_candidates entry."""
+    blocks = [
+        {"block_id": "c1", "page": 5, "role": "figure_caption",
+         "text": "Figure 1. Only caption", "bbox": [100, 100, 300, 130]},
+        {"block_id": "a1", "page": 5, "role": "figure_asset",
+         "bbox": [50, 0, 250, 90]},
+    ]
+    corpus = FigureCorpus.from_blocks(blocks, page_width=1200)
+    index = FigureCandidateIndex.from_corpus(corpus)
+
+    assert 5 not in index.sidecar_candidates
+
+    state = FigurePipelineState(corpus=corpus, candidate_index=index, ledger=OwnershipLedger())
+    report = SidecarPass().run(state)
+
+    assert len(report.accepted) == 0
+
+
+def test_sidecar_pass_does_not_steal_assets_from_protected_same_page_matches():
+    """Two narrow captions on page 5; c1 already has a protected same-page match.
+
+    SidecarPass should skip c1 and match only c2.
+    """
+    blocks = [
+        {"block_id": "c1", "page": 5, "role": "figure_caption",
+         "text": "Figure 1. Protected", "bbox": [100, 100, 300, 130]},
+        {"block_id": "c2", "page": 5, "role": "figure_caption",
+         "text": "Figure 2. Available", "bbox": [100, 300, 310, 330]},
+        {"block_id": "a1", "page": 5, "role": "figure_asset",
+         "bbox": [50, 0, 250, 90]},
+        {"block_id": "a2", "page": 5, "role": "figure_asset",
+         "bbox": [50, 180, 260, 270]},
+    ]
+    corpus = FigureCorpus.from_blocks(blocks, page_width=1200)
+    index = FigureCandidateIndex.from_corpus(corpus)
+
+    assert 5 in index.sidecar_candidates
+    assert len(index.sidecar_candidates[5]) == 2
+
+    state = FigurePipelineState(corpus=corpus, candidate_index=index, ledger=OwnershipLedger())
+
+    # Pre-populate a protected same-page match for c1
+    c1_ref = ResourceRef(kind="legend", page=5, block_id="c1", figure_no=1)
+    a1_ref = ResourceRef(kind="asset", page=5, block_id="a1")
+    state.ledger.try_claim_assets([a1_ref], owner=c1_ref, reason="same_page_primary")
+    state.matches.append({
+        "legend_block_id": "c1",
+        "settlement_type": "same_page",
+        "matched_assets": [{"block_id": "a1"}],
+        "flags": [],
+        "figure_number": 1,
+    })
+
+    report = SidecarPass().run(state)
+
+    # c1 should be skipped (protected), c2 should get a sidecar match
+    sidecar_matches = [m for m in state.matches if m.get("settlement_type") == "sidecar"]
+    assert len(sidecar_matches) == 1
+    assert sidecar_matches[0]["legend_block_id"] == "c2"
+    assert len(sidecar_matches[0].get("matched_assets", [])) >= 1