Optimizing Camelot for Complex Insurance Tables

This walkthrough extends the Table Parsing with Camelot stage with the page-chunking, memory-guarded isolation, deterministic flavor routing, and merged-cell reconstruction needed to parse high-page-count commercial declarations and loss-run schedules at portfolio scale, and it sits inside the broader Policy PDF Parsing & Extraction Workflows architecture.

A naive camelot.read_pdf(path, pages="all") call works on a single-page personal-auto declarations sheet and detonates on a 180-page commercial-lines policy. Camelot loads the whole file into RAM, shells out to Ghostscript to rasterize every page, and builds OpenCV line-detection matrices before it yields a single DataFrame. Across a batch of multi-state riders and scanned loss runs, three concrete failures dominate: the worker process is killed by the OOM reaper mid-document, the lattice flavor finds no grid on a borderless premium schedule and silently shifts a deductible into the limit column, and a merged “Total Insured Value” cell fragments into duplicate row indices that corrupt every downstream rating record. This page resolves all three by isolating extraction to bounded page ranges under a hard memory ceiling, routing flavor selection on a measured accuracy score rather than a hardcoded guess, and consolidating spanned cells before the schema gate ever sees them.

Camelot’s output is governed by native dependencies, so pin the entire stack — an unpinned Ghostscript or OpenCV bump silently changes lattice line detection and breaks reproducibility across deploys. Confirm the page has a vector layer first; image-only scans carry no strokes for Camelot to read and must be diverted to the OCR Integration & Sync pathway before they reach this code.

Copy
python==3.11
camelot-py[cv]==0.11.0
ghostscript==10.02.1   # system package; pin in your container image
opencv-python-headless==4.9.0.80
pandas==2.2.2
pydantic==2.7.1
psutil==5.9.8

Copy
# Verify the native toolchain is present before any extraction runs
gs --version            # must print 10.02.x
python -c "import camelot, cv2; print(camelot.__version__, cv2.__version__)"

Never hand Camelot the whole document. Resolve the page count once, slice it into bounded ranges, and run each slice inside a guard that samples RSS and raises an explicit, catchable error the instant the worker approaches its container limit. The guard converts a silent OOM kill into a routable event.

Copy
from __future__ import annotations

import gc
import logging
from dataclasses import dataclass

import camelot
import pandas as pd
import psutil
from pypdf import PdfReader

logger = logging.getLogger("camelot.extract")


class MemoryCeilingExceeded(RuntimeError):
    """Raised when a worker nears its container memory limit mid-extraction."""


@dataclass(frozen=True)
class ChunkConfig:
    pages_per_chunk: int = 10
    rss_ceiling_mb: int = 512


def page_ranges(pdf_path: str, cfg: ChunkConfig) -> list[str]:
    total = len(PdfReader(pdf_path).pages)
    return [
        f"{start + 1}-{min(start + cfg.pages_per_chunk, total)}"
        for start in range(0, total, cfg.pages_per_chunk)
    ]


def extract_chunk(pdf_path: str, page_range: str, cfg: ChunkConfig) -> list[pd.DataFrame]:
    rss_mb = psutil.Process().memory_info().rss / (1024 * 1024)
    if rss_mb > cfg.rss_ceiling_mb:
        raise MemoryCeilingExceeded(f"RSS {rss_mb:.0f}MB before pages {page_range}")

    tables = camelot.read_pdf(
        pdf_path,
        pages=page_range,
        flavor="lattice",
        process_background=False,
        split_text=True,
    )
    frames = [t.df.copy() for t in tables]
    logger.info("extracted", extra={"pages": page_range, "tables": len(frames)})

    del tables          # drop Camelot's native handles promptly
    gc.collect()
    return frames

lattice and stream are not interchangeable: lattice needs ruled grid lines, stream needs stable whitespace columns. Hardcoding either degrades silently on every document that violates its assumption. Run lattice first, read the per-table accuracy out of parsing_report, and only fall through to stream — with carrier-calibrated columns and table_areas — when the score drops below threshold. If both flavors stay low, raise so the document leaves the fast path instead of poisoning the canonical store.

Copy
from dataclasses import dataclass, field


class LowConfidenceTable(RuntimeError):
    """Both flavors failed to reconstruct the grid above threshold."""


@dataclass(frozen=True)
class FlavorConfig:
    min_accuracy: float = 90.0
    stream_columns: dict[str, str] = field(default_factory=dict)  # carrier_id -> "x1,x2,..."
    stream_areas: dict[str, str] = field(default_factory=dict)    # carrier_id -> "x1,y1,x2,y2"


def best_table(pdf_path: str, page_range: str, carrier_id: str, cfg: FlavorConfig) -> pd.DataFrame:
    lattice = camelot.read_pdf(pdf_path, pages=page_range, flavor="lattice", split_text=True)
    for table in lattice:
        if table.parsing_report["accuracy"] >= cfg.min_accuracy:
            return table.df.copy()

    columns = cfg.stream_columns.get(carrier_id)
    areas = cfg.stream_areas.get(carrier_id)
    stream = camelot.read_pdf(
        pdf_path,
        pages=page_range,
        flavor="stream",
        columns=[columns] if columns else None,
        table_areas=[areas] if areas else None,
        split_text=True,
    )
    ranked = sorted(stream, key=lambda t: t.parsing_report["accuracy"], reverse=True)
    if ranked and ranked[0].parsing_report["accuracy"] >= cfg.min_accuracy:
        return ranked[0].df.copy()

    best = max(
        (t.parsing_report["accuracy"] for t in (*lattice, *stream)),
        default=0.0,
    )
    raise LowConfidenceTable(f"{carrier_id} pages {page_range}: best accuracy {best:.1f}")

A merged “Total Insured Value” header or a two-line address splits across phantom rows that share the same vertical band. Group adjacent fragments by their row position, coalesce them, then hand the clean frame to a strict schema gate. Anything missing a mandatory field is rejected to a quarantine queue rather than persisted, and the column-normalization that follows belongs to Field Mapping Strategies.

Copy
from decimal import Decimal, InvalidOperation

from pydantic import BaseModel, field_validator


class ScheduleRow(BaseModel):
    coverage: str
    limit: Decimal
    deductible: Decimal
    premium: Decimal

    @field_validator("limit", "deductible", "premium", mode="before")
    @classmethod
    def parse_currency(cls, raw: str) -> Decimal:
        try:
            return Decimal(str(raw).replace("$", "").replace(",", "").strip() or "0")
        except InvalidOperation as exc:
            raise ValueError(f"unparseable currency: {raw!r}") from exc


def consolidate_merged_rows(df: pd.DataFrame) -> pd.DataFrame:
    """Coalesce fragments where a spanned cell left blank leading columns."""
    rows: list[list[str]] = []
    for _, cells in df.iterrows():
        values = [c.replace("\n", " ").strip() for c in cells.tolist()]
        if rows and values[0] == "":  # continuation of the previous spanned row
            rows[-1] = [(a + " " + b).strip() for a, b in zip(rows[-1], values)]
        else:
            rows.append(values)
    return pd.DataFrame(rows[1:], columns=rows[0])


def validate_rows(df: pd.DataFrame) -> tuple[list[ScheduleRow], list[dict]]:
    valid, quarantined = [], []
    for record in df.to_dict("records"):
        try:
            valid.append(ScheduleRow(**record))
        except ValueError as exc:
            quarantined.append({"record": record, "error": str(exc)})
    return valid, quarantined

Confirm correctness on a representative fixture before trusting the pipeline on live carrier feeds. Assert that chunking never exceeds the ceiling, that flavor routing recovers a known-good schedule above threshold, and that a deliberately merged row reconstructs to the expected column count.

Copy
def test_merged_total_insured_value_reconstructs() -> None:
    raw = pd.DataFrame(
        [
            ["Coverage", "Limit", "Deductible", "Premium"],
            ["Building", "$1,500,000", "$10,000", "$8,420"],
            ["", "and Contents", "", ""],            # spanned continuation
            ["Business Income", "$250,000", "$5,000", "$2,110"],
        ]
    )
    clean = consolidate_merged_rows(raw)
    valid, quarantined = validate_rows(clean)

    assert not quarantined
    assert {r.coverage for r in valid} == {"Building and Contents", "Business Income"}
    assert valid[0].limit == Decimal("1500000")
    assert valid[0].deductible == Decimal("10000")


def test_chunks_cover_every_page() -> None:
    cfg = ChunkConfig(pages_per_chunk=10)
    ranges = ["1-10", "11-20", "21-23"]  # for a 23-page document
    spanned = [n for r in ranges for n in range(int(r.split("-")[0]), int(r.split("-")[1]) + 1)]
    assert spanned == list(range(1, 24))

In production, run a per-document invariant: the count of validated ScheduleRow objects plus quarantined records must equal the raw row count, so no row is ever dropped without a trace. Surface flavor-routing accuracy and quarantine rate as metrics and alert when either drifts from its carrier baseline.

Every extracted schedule row must trace back to the exact page and parameters that produced it. Persist the source PDF’s SHA-256 hash, the resolved page_range, the chosen flavor, the parsing_report accuracy, and the pinned Camelot and Ghostscript versions alongside each validated row, and write those records append-only. That lineage lets the Coverage Validation Rules engine cross-check a recovered limit against the bound policy on file, and it satisfies NAIC data-governance expectations and state department-of-insurance examination requests by letting an examiner reconstruct precisely how $1,500,000 was derived from a specific page under a specific configuration — without exposing the production database.

Each bounded page range runs through the same decision path: a memory guard, a lattice attempt scored against threshold, a stream fall-through with carrier-calibrated columns, and three terminal lanes — the OOM guard, the validated store, and the quarantine/OCR diversion.

• Table Parsing with Camelot — the parent stage this walkthrough extends
• Extracting Coverage Limits from Scanned Policy PDFs — coordinate anchoring when there is no grid to parse
• Handling Rotated Pages in Policy Documents — deskew scans before they reach the optical fallback
• Building Async Batch Processors for Daily Policy Ingestion — run this extractor across a portfolio under bounded concurrency
• Validating Deductible Thresholds Automatically — check recovered limits and deductibles against the bound policy