feat: text-layer grounding (extractor authority, guard rescue tier, verifier oracle + hi-DPI crops)
- backend/text_layer.py: PyMuPDF text-layer extraction, fuzzy evidence
bbox matching, 300-DPI crop rendering, coverage-gap signal
- extractor (classic + agent): TEXT LAYER block appended at call sites;
grounding guard gains text-layer rescue tier (grounding=text_layer stamp)
- verifier: {text_layer} oracle excerpt + evidence-located hi-DPI crops
replacing full-page images (fallback preserved, I2 guard intact)
- coverage gaps: text-bearing pages with zero extraction -> failed-scope
gap findings (agent) / log-only (classic)
- config knobs: TEXT_LAYER_ENABLED/MIN_CHARS/MAX_CHARS, VERIFY_TEXT_MAX_CHARS,
VERIFY_HI_DPI_CROPS, VERIFY_CROP_DPI, VERIFY_CROP_MARGIN_PTS
- tests: 22 new (text_layer unit, grounding/render, runner-level flow)
Spec: docs/superpowers/specs/2026-08-12-text-layer-grounding-design.md
This commit is contained in:
@@ -0,0 +1,230 @@
|
||||
"""
|
||||
text_layer.py - deterministic PDF text-layer extraction (PyMuPDF, no LLM).
|
||||
|
||||
Most CAD-produced drawing sets carry a real vector text layer. We extract it
|
||||
once per job and feed it to the extractor (grounding), the grounding guard
|
||||
(rescue tier), and the wave-5b verifier (text oracle + high-DPI evidence
|
||||
crops). Pages below TEXT_LAYER_MIN_CHARS of text are treated as having no
|
||||
text layer (scanned/raster sheets stay vision-only).
|
||||
|
||||
If PyMuPDF is unavailable the module degrades gracefully: every public
|
||||
function returns empty/None, equivalent to TEXT_LAYER_ENABLED=false.
|
||||
"""
|
||||
|
||||
import re
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
|
||||
from backend import config
|
||||
|
||||
try: # PyMuPDF >= 1.24 prefers the pymupdf name; fitz works everywhere.
|
||||
import pymupdf as fitz
|
||||
except ImportError: # pragma: no cover - older PyMuPDF
|
||||
try:
|
||||
import fitz
|
||||
except ImportError: # pragma: no cover - PyMuPDF not installed
|
||||
fitz = None
|
||||
|
||||
_warned_unavailable = False
|
||||
|
||||
# Word token normalization for evidence matching: lowercase alphanumeric only.
|
||||
_TOKEN_RE = re.compile(r"[^a-z0-9]+")
|
||||
# Fuzzy match floor: fraction of needle tokens that must align with the page's
|
||||
# word sequence for a bbox to count as a confident evidence location.
|
||||
_FUZZY_MIN_RATIO = 0.6
|
||||
|
||||
|
||||
def _fitz_or_none():
|
||||
"""Return the fitz module, logging once if PyMuPDF is missing."""
|
||||
global _warned_unavailable
|
||||
if fitz is None and not _warned_unavailable:
|
||||
print("[TextLayer] PyMuPDF not available - text-layer grounding disabled")
|
||||
_warned_unavailable = True
|
||||
return fitz
|
||||
|
||||
|
||||
def extract_text_layers(pdf_path: str) -> Dict[int, Dict]:
|
||||
"""
|
||||
Extract the text layer of every page. Returns {1-based page_number:
|
||||
{"text": str, "words": [{"text", "bbox": (x0,y0,x1,y1)}, ...],
|
||||
"has_text_layer": bool}}. Returns {} when disabled or unavailable.
|
||||
"""
|
||||
if not config.TEXT_LAYER_ENABLED:
|
||||
return {}
|
||||
f = _fitz_or_none()
|
||||
if f is None:
|
||||
return {}
|
||||
try:
|
||||
doc = f.open(pdf_path)
|
||||
except Exception as exc:
|
||||
print(f"[TextLayer] could not open {pdf_path}: {exc}")
|
||||
return {}
|
||||
layers: Dict[int, Dict] = {}
|
||||
try:
|
||||
for index in range(doc.page_count):
|
||||
page = doc[index]
|
||||
text = page.get_text("text") or ""
|
||||
words = [
|
||||
{"text": w[4], "bbox": (w[0], w[1], w[2], w[3])}
|
||||
for w in (page.get_text("words") or [])
|
||||
]
|
||||
has_text_layer = len(text.strip()) >= config.TEXT_LAYER_MIN_CHARS
|
||||
if not has_text_layer:
|
||||
print(f"[TextLayer] Page {index + 1}: {len(text.strip())} chars "
|
||||
f"(< TEXT_LAYER_MIN_CHARS={config.TEXT_LAYER_MIN_CHARS}) - "
|
||||
f"vision-only")
|
||||
layers[index + 1] = {
|
||||
"text": text,
|
||||
"words": words,
|
||||
"has_text_layer": has_text_layer,
|
||||
}
|
||||
finally:
|
||||
doc.close()
|
||||
return layers
|
||||
|
||||
|
||||
def attach_text_layers(
|
||||
pdf_path: str,
|
||||
pages: List[Dict],
|
||||
text_dir: Optional[str] = None,
|
||||
) -> Dict[int, List[Dict]]:
|
||||
"""
|
||||
Attach page["text_layer"] (text or None) to each converted page dict and
|
||||
return the runner-local {page_number: words} map (kept off page dicts -
|
||||
those get serialized). When text_dir is set, dump one .txt per page there
|
||||
(plain file writes; ProjectMemory is a closed registry).
|
||||
"""
|
||||
layers = extract_text_layers(pdf_path)
|
||||
page_words: Dict[int, List[Dict]] = {}
|
||||
for page in pages:
|
||||
layer = layers.get(page["page_number"]) or {}
|
||||
page["text_layer"] = layer.get("text") if layer.get("has_text_layer") else None
|
||||
page_words[page["page_number"]] = layer.get("words") or []
|
||||
if text_dir and layers:
|
||||
import os
|
||||
os.makedirs(text_dir, exist_ok=True)
|
||||
for page_number, layer in layers.items():
|
||||
if not layer.get("has_text_layer"):
|
||||
continue
|
||||
with open(os.path.join(text_dir, f"page-{page_number:03d}.txt"),
|
||||
"w", encoding="utf-8") as fh:
|
||||
fh.write(layer.get("text") or "")
|
||||
return page_words
|
||||
|
||||
|
||||
def _tokens(text: str) -> List[str]:
|
||||
return [t for t in _TOKEN_RE.split(text.lower()) if t]
|
||||
|
||||
|
||||
def _union_bbox(boxes: List[Tuple[float, float, float, float]]):
|
||||
return (
|
||||
min(b[0] for b in boxes),
|
||||
min(b[1] for b in boxes),
|
||||
max(b[2] for b in boxes),
|
||||
max(b[3] for b in boxes),
|
||||
)
|
||||
|
||||
|
||||
def find_evidence_bbox(
|
||||
words: List[Dict],
|
||||
needle: str,
|
||||
) -> Optional[Tuple[float, float, float, float]]:
|
||||
"""
|
||||
Best-effort fuzzy substring match of an evidence source_text against the
|
||||
page's word sequence. Returns the union bbox of the matched words, or
|
||||
None when nothing aligns confidently.
|
||||
|
||||
Exact contiguous token runs win; otherwise the best-scoring window with
|
||||
>= _FUZZY_MIN_RATIO token alignment is accepted (vision quotes imperfectly
|
||||
but the value is real page text).
|
||||
"""
|
||||
if not words or not needle:
|
||||
return None
|
||||
needle_tokens = _tokens(str(needle))
|
||||
if not needle_tokens:
|
||||
return None
|
||||
page_tokens = [_tokens(w.get("text") or "") for w in words]
|
||||
# Flatten multi-token words, remembering which word each token came from.
|
||||
flat: List[Tuple[str, int]] = []
|
||||
for word_index, parts in enumerate(page_tokens):
|
||||
for part in parts:
|
||||
flat.append((part, word_index))
|
||||
if not flat:
|
||||
return None
|
||||
|
||||
n = len(needle_tokens)
|
||||
best_span = None
|
||||
best_score = 0.0
|
||||
for start in range(0, len(flat)):
|
||||
window = flat[start:start + n]
|
||||
if not window:
|
||||
break
|
||||
score = sum(1 for i, tok in enumerate(needle_tokens)
|
||||
if i < len(window) and window[i][0] == tok) / n
|
||||
if score > best_score:
|
||||
best_score = score
|
||||
best_span = window
|
||||
if best_score == 1.0:
|
||||
break
|
||||
if best_span is None or best_score < _FUZZY_MIN_RATIO:
|
||||
return None
|
||||
word_indexes = {word_index for _, word_index in best_span}
|
||||
return _union_bbox([words[i]["bbox"] for i in sorted(word_indexes)])
|
||||
|
||||
|
||||
def render_crop(
|
||||
pdf_path: str,
|
||||
page_number: int,
|
||||
bbox: Tuple[float, float, float, float],
|
||||
dpi: Optional[int] = None,
|
||||
margin_pts: Optional[float] = None,
|
||||
) -> Optional[bytes]:
|
||||
"""
|
||||
Render a clip of one page around bbox (+ margin, clamped to the page) at
|
||||
the given DPI and return JPEG bytes, or None on any failure.
|
||||
"""
|
||||
f = _fitz_or_none()
|
||||
if f is None:
|
||||
return None
|
||||
dpi = dpi or config.VERIFY_CROP_DPI
|
||||
margin_pts = config.VERIFY_CROP_MARGIN_PTS if margin_pts is None else margin_pts
|
||||
try:
|
||||
doc = f.open(pdf_path)
|
||||
try:
|
||||
page = doc[page_number - 1]
|
||||
rect = f.Rect(
|
||||
bbox[0] - margin_pts,
|
||||
bbox[1] - margin_pts,
|
||||
bbox[2] + margin_pts,
|
||||
bbox[3] + margin_pts,
|
||||
) & page.rect
|
||||
if rect.is_empty:
|
||||
return None
|
||||
pix = page.get_pixmap(clip=rect, dpi=dpi)
|
||||
return pix.tobytes("jpeg")
|
||||
finally:
|
||||
doc.close()
|
||||
except Exception as exc:
|
||||
print(f"[TextLayer] render_crop failed on page {page_number}: {exc}")
|
||||
return None
|
||||
|
||||
|
||||
def coverage_gaps(pages: List[Dict], sheets: List[Dict]) -> List[int]:
|
||||
"""
|
||||
Page numbers that have a text layer but whose extraction failed or
|
||||
returned 0 objects - the silent extraction-loss signal. Logs one
|
||||
[TextLayer] line per gap.
|
||||
"""
|
||||
by_page = {s.get("page_number"): s for s in sheets or []}
|
||||
gaps: List[int] = []
|
||||
for page in pages:
|
||||
text = page.get("text_layer")
|
||||
if not text:
|
||||
continue
|
||||
sheet = by_page.get(page["page_number"])
|
||||
extracted = len(sheet.get("assertions") or []) if sheet else 0
|
||||
if extracted == 0:
|
||||
gaps.append(page["page_number"])
|
||||
print(f"[TextLayer] Page {page['page_number']}: text layer present "
|
||||
f"({len(text)} chars) but no objects extracted — possible "
|
||||
f"extraction gap")
|
||||
return gaps
|
||||
Reference in New Issue
Block a user