20258147097

Committed 16 Dec 2025 05:59AM UTC coverage: 89.094% (+0.4%) from 88.668%

Build # 20258147097

Build Type

push

github

Committed by

bramp

Commit Message

chore: Minor cleanups

- Add TODO comment about BatchClassificationResult naming
- Remove completed testing improvements from TODO.md

Coverage Stats

12818 of 14387 relevant lines covered (89.09%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

99.12

/src/build_a_long/pdf_extract/classifier/classifier.py

"""
Rule-based classifier for labeling page elements.

Pipeline order and dependencies
--------------------------------
The classification pipeline operates in two main phases:

1. **Bottom-up Scoring**: All classifiers run independently to identify potential
   candidates (e.g. page numbers, part counts, step numbers) and score them based
   on heuristics. No construction of final elements happens here.

2. **Top-down Construction**: The root `PageClassifier` is invoked to construct
   the final `Page` object. It recursively requests the construction of its
   dependencies (e.g. "Give me the best PageNumber"), which in turn construct
   their own dependencies. This ensures a consistent and validated object tree.

"""

from __future__ import annotations

import logging

from build_a_long.pdf_extract.classifier.bags import (
    BagNumberClassifier,
    LoosePartSymbolClassifier,
    OpenBagClassifier,
)
from build_a_long.pdf_extract.classifier.batch_classification_result import (
    BatchClassificationResult,
)
from build_a_long.pdf_extract.classifier.block_filter import (
    filter_duplicate_blocks,
    filter_overlapping_text_blocks,
)
from build_a_long.pdf_extract.classifier.classification_result import (
    ClassificationResult,
)
from build_a_long.pdf_extract.classifier.classifier_config import ClassifierConfig
from build_a_long.pdf_extract.classifier.pages import (
    PageHintCollection,
)
from build_a_long.pdf_extract.classifier.pages.background_classifier import (
    BackgroundClassifier,
)
from build_a_long.pdf_extract.classifier.pages.divider_classifier import (
    DividerClassifier,
)
from build_a_long.pdf_extract.classifier.pages.page_classifier import PageClassifier
from build_a_long.pdf_extract.classifier.pages.page_number_classifier import (
    PageNumberClassifier,
)
from build_a_long.pdf_extract.classifier.pages.preview_classifier import (
    PreviewClassifier,
)
from build_a_long.pdf_extract.classifier.pages.progress_bar_classifier import (
    ProgressBarClassifier,
)
from build_a_long.pdf_extract.classifier.pages.progress_bar_indicator_classifier import (
    ProgressBarIndicatorClassifier,
)
from build_a_long.pdf_extract.classifier.pages.trivia_text_classifier import (
    TriviaTextClassifier,
)
from build_a_long.pdf_extract.classifier.parts import (
    PartCountClassifier,
    PartNumberClassifier,
    PartsClassifier,
    PartsImageClassifier,
    PartsListClassifier,
    PieceLengthClassifier,
    ScaleClassifier,
    ShineClassifier,
)
from build_a_long.pdf_extract.classifier.removal_reason import RemovalReason
from build_a_long.pdf_extract.classifier.steps import (
    ArrowClassifier,
    DiagramClassifier,
    RotationSymbolClassifier,
    StepClassifier,
    StepCountClassifier,
    StepNumberClassifier,
    SubAssemblyClassifier,
    SubStepClassifier,
    SubStepNumberClassifier,
)
from build_a_long.pdf_extract.classifier.text import FontSizeHints, TextHistogram
from build_a_long.pdf_extract.classifier.topological_sort import topological_sort
from build_a_long.pdf_extract.extractor import PageData
from build_a_long.pdf_extract.extractor.bbox import filter_contained
from build_a_long.pdf_extract.extractor.lego_page_elements import (
    PageNumber,
    PartCount,
    PartsList,
    StepNumber,
)
from build_a_long.pdf_extract.extractor.page_blocks import Blocks

logger = logging.getLogger(__name__)

# Pages with more blocks than this threshold will be skipped during classification.
# This avoids O(n²) algorithms (like duplicate detection) that become prohibitively
# slow on pages with thousands of vector drawings. Such pages are typically info
# pages where each character is a separate vector graphic.
# TODO: Add spatial indexing to handle high-block pages efficiently.
MAX_BLOCKS_PER_PAGE = 1000


# TODO require config, so we don't accidentally use default empty config
def classify_elements(
    page: PageData, config: ClassifierConfig | None = None
) -> ClassificationResult:
    """Classify and label elements on a single page using rule-based heuristics.

    Args:
        page: A single PageData object to classify.
        config: Optional classifier configuration with font/page hints.
            If None, uses default empty configuration (no hints).
            For better classification accuracy, pass a config with
            FontSizeHints computed from multiple pages of the same PDF.

    Returns:
        A ClassificationResult object containing the classification results.
    """
    if config is None:
        config = ClassifierConfig()
    classifier = Classifier(config)

    return classifier.classify(page)


def classify_pages(
    pages: list[PageData], pages_for_hints: list[PageData] | None = None
) -> BatchClassificationResult:
    """Classify and label elements across multiple pages using rule-based heuristics.

    This function performs a three-phase process:
    1. Filtering phase: Mark duplicate/similar blocks as removed on each page
    2. Analysis phase: Build font size hints from text properties (excluding
       removed blocks)
    3. Classification phase: Use hints to guide element classification

    Args:
        pages: A list of PageData objects to classify.
        pages_for_hints: Optional list of pages to use for generating font/page hints.
            If None, uses `pages`. This allows generating hints from all pages
            while only classifying a subset (e.g., when using --pages filter).

    Returns:
        BatchClassificationResult containing per-page results and global histogram
    """

    # TODO There is a bunch of duplication in here between hints and non-hints. Refactor

    # Use all pages for hint generation if provided, otherwise use selected pages
    hint_pages = pages_for_hints if pages_for_hints is not None else pages

    # Phase 1: Filter duplicate blocks on each page and track removals
    # Skip pages with too many blocks to avoid O(n²) performance issues
    removed_blocks_per_page: list[dict[Blocks, RemovalReason]] = []
    skipped_pages: set[int] = set()  # Track page numbers that are skipped

    for page_data in pages:
        # Skip pages with too many blocks - these are likely info/inventory pages
        # with vectorized text that cause O(n²) algorithms to be very slow
        if len(page_data.blocks) > MAX_BLOCKS_PER_PAGE:
            logger.debug(
                f"Page {page_data.page_number}: skipping classification "
                f"({len(page_data.blocks)} blocks exceeds threshold of "
                f"{MAX_BLOCKS_PER_PAGE})"
            )
            skipped_pages.add(page_data.page_number)
            removed_blocks_per_page.append({})
            continue

        kept_blocks = page_data.blocks

        # Filter overlapping text blocks (e.g., "4" and "43" at same origin)
        kept_blocks, text_removed = filter_overlapping_text_blocks(kept_blocks)

        # Filter duplicate image/drawing blocks based on IOU
        kept_blocks, bbox_removed = filter_duplicate_blocks(kept_blocks)

        # Combine all removal mappings into a single dict for this page
        combined_removed_mapping = {
            **text_removed,
            **bbox_removed,
        }

        logger.debug(
            f"Page {page_data.page_number}: "
            f"filtered {len(text_removed)} overlapping text, "
            f"{len(bbox_removed)} duplicate bbox blocks"
        )

        removed_blocks_per_page.append(combined_removed_mapping)

    # Phase 2: Extract font size hints from hint pages (excluding removed blocks)
    # Build pages with non-removed blocks for hint extraction and histogram

    # Filter duplicates from hint pages (may be different from pages to classify)
    hint_pages_without_duplicates = []
    for page_data in hint_pages:
        # Skip high-block pages for hints too (same threshold)
        if len(page_data.blocks) > MAX_BLOCKS_PER_PAGE:
            continue

        # TODO We are re-filtering duplicates here; optimize by changing the API
        # to accept one list of PageData, and seperate by page_numbers.
        kept_blocks = page_data.blocks
        kept_blocks, _ = filter_overlapping_text_blocks(kept_blocks)
        kept_blocks, _ = filter_duplicate_blocks(kept_blocks)

        hint_pages_without_duplicates.append(
            PageData(
                page_number=page_data.page_number,
                bbox=page_data.bbox,
                blocks=kept_blocks,
            )
        )

    # Build pages without duplicates for classification
    pages_without_duplicates = []
    for page_data, removed_mapping in zip(pages, removed_blocks_per_page, strict=True):
        # We need to filter blocks that were removed by ANY filter
        non_removed_blocks = [
            block for block in page_data.blocks if block not in removed_mapping
        ]
        pages_without_duplicates.append(
            PageData(
                page_number=page_data.page_number,
                bbox=page_data.bbox,
                blocks=non_removed_blocks,
            )
        )

    # Generate hints from hint pages, histogram from pages to classify
    font_size_hints = FontSizeHints.from_pages(hint_pages_without_duplicates)
    page_hints = PageHintCollection.from_pages(hint_pages_without_duplicates)
    histogram = TextHistogram.from_pages(pages_without_duplicates)

    # Phase 3: Classify using the hints (on pages without duplicates)
    config = ClassifierConfig(font_size_hints=font_size_hints, page_hints=page_hints)
    classifier = Classifier(config)

    results = []
    for page_data, page_without_duplicates, removed_mapping in zip(
        pages, pages_without_duplicates, removed_blocks_per_page, strict=True
    ):
        # Handle skipped pages
        if page_data.page_number in skipped_pages:
            result = ClassificationResult(
                page_data=page_data,
                skipped_reason=(
                    f"Page has {len(page_data.blocks)} blocks, which exceeds "
                    f"the threshold of {MAX_BLOCKS_PER_PAGE}. This is likely an "
                    f"info/inventory page with vectorized text."
                ),
            )
            results.append(result)
            continue

        # Classify using only non-removed blocks
        result = classifier.classify(page_without_duplicates)

        # Update result to use original page_data (with all blocks)
        result.page_data = page_data

        # Mark removed blocks
        for removed_block, removal_reason in removed_mapping.items():
            result.mark_removed(removed_block, removal_reason)

        results.append(result)

    return BatchClassificationResult(results=results, histogram=histogram)


type Classifiers = (
    PageNumberClassifier
    | ProgressBarClassifier
    | ProgressBarIndicatorClassifier
    | PreviewClassifier
    | BackgroundClassifier
    | DividerClassifier
    | BagNumberClassifier
    | PartCountClassifier
    | PartNumberClassifier
    | StepNumberClassifier
    | StepCountClassifier
    | PieceLengthClassifier
    | ScaleClassifier
    | PartsClassifier
    | PartsListClassifier
    | PartsImageClassifier
    | ShineClassifier
    | OpenBagClassifier
    | LoosePartSymbolClassifier
    | DiagramClassifier
    | ArrowClassifier
    | SubAssemblyClassifier
    | StepClassifier
    | TriviaTextClassifier
    | PageClassifier
)


class Classifier:
    """
    Performs a single run of classification based on rules, configuration, and hints.
    This class should be stateless.
    """

    def __init__(self, config: ClassifierConfig):
        self.config = config
        # Sort classifiers topologically based on their dependencies
        self.classifiers = topological_sort(
            [
                PageNumberClassifier(config=config),
                ProgressBarIndicatorClassifier(config=config),
                ProgressBarClassifier(config=config),
                BackgroundClassifier(config=config),
                DividerClassifier(config=config),
                BagNumberClassifier(config=config),
                PartCountClassifier(config=config),
                PartNumberClassifier(config=config),
                StepNumberClassifier(config=config),
                SubStepNumberClassifier(config=config),
                StepCountClassifier(config=config),
                PieceLengthClassifier(config=config),
                ScaleClassifier(config=config),
                PartsClassifier(config=config),
                PartsListClassifier(config=config),
                DiagramClassifier(config=config),
                RotationSymbolClassifier(config=config),
                ArrowClassifier(config=config),
                PartsImageClassifier(config=config),
                ShineClassifier(config=config),
                OpenBagClassifier(config=config),
                LoosePartSymbolClassifier(config=config),
                PreviewClassifier(config=config),
                SubStepClassifier(config=config),
                SubAssemblyClassifier(config=config),
                StepClassifier(config=config),
                TriviaTextClassifier(config=config),
                PageClassifier(config=config),
            ]
        )

    def classify(self, page_data: PageData) -> ClassificationResult:
        """
        Runs the classification logic and returns a result.
        It does NOT modify page_data directly.

        The classification process runs in three phases:
        1. Score all classifiers (bottom-up) - auto-registers classifiers
        2. Construct final elements (top-down starting from Page)
        """
        result = ClassificationResult(page_data=page_data)

        logger.debug(f"Starting classification for page {page_data.page_number}")

        # 1. Score all classifiers (Bottom-Up)
        # Note: score() automatically registers each classifier for its output labels
        for classifier in self.classifiers:
            classifier.score(result)

        # 2. Construct (Top-Down)
        # Find the PageClassifier to start the construction process
        page_classifier = next(
            c for c in self.classifiers if isinstance(c, PageClassifier)
        )
        page_classifier.build_all(result)

        # 3. Validate that all page elements are tracked via candidates
        # This catches programming errors where elements are created directly
        # instead of via result.build()
        from build_a_long.pdf_extract.validation.rules import (
            assert_page_elements_tracked,
        )

        assert_page_elements_tracked(result)

        # TODO Do we actualy ever add warnings?
        warnings = self._log_post_classification_warnings(page_data, result)
        for warning in warnings:
            result.add_warning(warning)

        return result

    def _log_post_classification_warnings(
        self, page_data: PageData, result: ClassificationResult
    ) -> list[str]:
        warnings = []

        # Check if there's a page number
        page_numbers = result.get_winners_by_score("page_number", PageNumber)
        if not page_numbers:
            warnings.append(f"Page {page_data.page_number}: missing page number")

        # Get elements by label
        parts_lists = result.get_winners_by_score("parts_list", PartsList)
        part_counts = result.get_winners_by_score("part_count", PartCount)

        for pl in parts_lists:
            inside_counts = filter_contained(part_counts, pl.bbox)
            if not inside_counts:
                warnings.append(
                    f"Page {page_data.page_number}: parts list at {pl.bbox} "
                    f"contains no part counts"
                )

        steps = result.get_winners_by_score("step_number", StepNumber)
        ABOVE_EPS = 2.0
        for step in steps:
            sb = step.bbox
            above = [pl for pl in parts_lists if pl.bbox.y1 <= sb.y0 + ABOVE_EPS]
            if not above:
                warnings.append(
                    f"Page {page_data.page_number}: step number '{step.value}' "
                    f"at {sb} has no parts list above it"
                )
        return warnings

1	"""
2	Rule-based classifier for labeling page elements.
3
4	Pipeline order and dependencies
5	--------------------------------
6	The classification pipeline operates in two main phases:
7
8	1. Bottom-up Scoring: All classifiers run independently to identify potential
9	candidates (e.g. page numbers, part counts, step numbers) and score them based
10	on heuristics. No construction of final elements happens here.
11
12	2. Top-down Construction: The root `PageClassifier` is invoked to construct
13	the final `Page` object. It recursively requests the construction of its
14	dependencies (e.g. "Give me the best PageNumber"), which in turn construct
15	their own dependencies. This ensures a consistent and validated object tree.
16
17	"""
18
19	from __future__ import annotations	1✔
20
21	import logging	1✔
22
23	from build_a_long.pdf_extract.classifier.bags import (	1✔
24	BagNumberClassifier,
25	LoosePartSymbolClassifier,
26	OpenBagClassifier,
27	)
28	from build_a_long.pdf_extract.classifier.batch_classification_result import (	1✔
29	BatchClassificationResult,
30	)
31	from build_a_long.pdf_extract.classifier.block_filter import (	1✔
32	filter_duplicate_blocks,
33	filter_overlapping_text_blocks,
34	)
35	from build_a_long.pdf_extract.classifier.classification_result import (	1✔
36	ClassificationResult,
37	)
38	from build_a_long.pdf_extract.classifier.classifier_config import ClassifierConfig	1✔
39	from build_a_long.pdf_extract.classifier.pages import (	1✔
40	PageHintCollection,
41	)
42	from build_a_long.pdf_extract.classifier.pages.background_classifier import (	1✔
43	BackgroundClassifier,
44	)
45	from build_a_long.pdf_extract.classifier.pages.divider_classifier import (	1✔
46	DividerClassifier,
47	)
48	from build_a_long.pdf_extract.classifier.pages.page_classifier import PageClassifier	1✔
49	from build_a_long.pdf_extract.classifier.pages.page_number_classifier import (	1✔
50	PageNumberClassifier,
51	)
52	from build_a_long.pdf_extract.classifier.pages.preview_classifier import (	1✔
53	PreviewClassifier,
54	)
55	from build_a_long.pdf_extract.classifier.pages.progress_bar_classifier import (	1✔
56	ProgressBarClassifier,
57	)
58	from build_a_long.pdf_extract.classifier.pages.progress_bar_indicator_classifier import (	1✔
59	ProgressBarIndicatorClassifier,
60	)
61	from build_a_long.pdf_extract.classifier.pages.trivia_text_classifier import (	1✔
62	TriviaTextClassifier,
63	)
64	from build_a_long.pdf_extract.classifier.parts import (	1✔
65	PartCountClassifier,
66	PartNumberClassifier,
67	PartsClassifier,
68	PartsImageClassifier,
69	PartsListClassifier,
70	PieceLengthClassifier,
71	ScaleClassifier,
72	ShineClassifier,
73	)
74	from build_a_long.pdf_extract.classifier.removal_reason import RemovalReason	1✔
75	from build_a_long.pdf_extract.classifier.steps import (	1✔
76	ArrowClassifier,
77	DiagramClassifier,
78	RotationSymbolClassifier,
79	StepClassifier,
80	StepCountClassifier,
81	StepNumberClassifier,
82	SubAssemblyClassifier,
83	SubStepClassifier,
84	SubStepNumberClassifier,
85	)
86	from build_a_long.pdf_extract.classifier.text import FontSizeHints, TextHistogram	1✔
87	from build_a_long.pdf_extract.classifier.topological_sort import topological_sort	1✔
88	from build_a_long.pdf_extract.extractor import PageData	1✔
89	from build_a_long.pdf_extract.extractor.bbox import filter_contained	1✔
90	from build_a_long.pdf_extract.extractor.lego_page_elements import (	1✔
91	PageNumber,
92	PartCount,
93	PartsList,
94	StepNumber,
95	)
96	from build_a_long.pdf_extract.extractor.page_blocks import Blocks	1✔
97
98	logger = logging.getLogger(__name__)	1✔
99
100	# Pages with more blocks than this threshold will be skipped during classification.
101	# This avoids O(n²) algorithms (like duplicate detection) that become prohibitively
102	# slow on pages with thousands of vector drawings. Such pages are typically info
103	# pages where each character is a separate vector graphic.
104	# TODO: Add spatial indexing to handle high-block pages efficiently.
105	MAX_BLOCKS_PER_PAGE = 1000	1✔
106
107
108	# TODO require config, so we don't accidentally use default empty config
109	def classify_elements(	1✔
110	page: PageData, config: ClassifierConfig \| None = None
111	) -> ClassificationResult:
112	"""Classify and label elements on a single page using rule-based heuristics.
113
114	Args:
115	page: A single PageData object to classify.
116	config: Optional classifier configuration with font/page hints.
117	If None, uses default empty configuration (no hints).
118	For better classification accuracy, pass a config with
119	FontSizeHints computed from multiple pages of the same PDF.
120
121	Returns:
122	A ClassificationResult object containing the classification results.
123	"""
124	if config is None:	1✔
125	config = ClassifierConfig()	1✔
126	classifier = Classifier(config)	1✔
127
128	return classifier.classify(page)	1✔
129
130
131	def classify_pages(	1✔
132	pages: list[PageData], pages_for_hints: list[PageData] \| None = None
133	) -> BatchClassificationResult:
134	"""Classify and label elements across multiple pages using rule-based heuristics.
135
136	This function performs a three-phase process:
137	1. Filtering phase: Mark duplicate/similar blocks as removed on each page
138	2. Analysis phase: Build font size hints from text properties (excluding
139	removed blocks)
140	3. Classification phase: Use hints to guide element classification
141
142	Args:
143	pages: A list of PageData objects to classify.
144	pages_for_hints: Optional list of pages to use for generating font/page hints.
145	If None, uses `pages`. This allows generating hints from all pages
146	while only classifying a subset (e.g., when using --pages filter).
147
148	Returns:
149	BatchClassificationResult containing per-page results and global histogram
150	"""
151
152	# TODO There is a bunch of duplication in here between hints and non-hints. Refactor
153
154	# Use all pages for hint generation if provided, otherwise use selected pages
155	hint_pages = pages_for_hints if pages_for_hints is not None else pages	1✔
156
157	# Phase 1: Filter duplicate blocks on each page and track removals
158	# Skip pages with too many blocks to avoid O(n²) performance issues
159	removed_blocks_per_page: list[dict[Blocks, RemovalReason]] = []	1✔
160	skipped_pages: set[int] = set() # Track page numbers that are skipped	1✔
161
162	for page_data in pages:	1✔
163	# Skip pages with too many blocks - these are likely info/inventory pages
164	# with vectorized text that cause O(n²) algorithms to be very slow
165	if len(page_data.blocks) > MAX_BLOCKS_PER_PAGE:	1✔
166	logger.debug(	1✔
167	f"Page {page_data.page_number}: skipping classification "
168	f"({len(page_data.blocks)} blocks exceeds threshold of "
169	f"{MAX_BLOCKS_PER_PAGE})"
170	)
171	skipped_pages.add(page_data.page_number)	1✔
172	removed_blocks_per_page.append({})	1✔
173	continue	1✔
174
175	kept_blocks = page_data.blocks	1✔
176
177	# Filter overlapping text blocks (e.g., "4" and "43" at same origin)
178	kept_blocks, text_removed = filter_overlapping_text_blocks(kept_blocks)	1✔
179
180	# Filter duplicate image/drawing blocks based on IOU
181	kept_blocks, bbox_removed = filter_duplicate_blocks(kept_blocks)	1✔
182
183	# Combine all removal mappings into a single dict for this page
184	combined_removed_mapping = {	1✔
185	**text_removed,
186	**bbox_removed,
187	}
188
189	logger.debug(	1✔
190	f"Page {page_data.page_number}: "
191	f"filtered {len(text_removed)} overlapping text, "
192	f"{len(bbox_removed)} duplicate bbox blocks"
193	)
194
195	removed_blocks_per_page.append(combined_removed_mapping)	1✔
196
197	# Phase 2: Extract font size hints from hint pages (excluding removed blocks)
198	# Build pages with non-removed blocks for hint extraction and histogram
199
200	# Filter duplicates from hint pages (may be different from pages to classify)
201	hint_pages_without_duplicates = []	1✔
202	for page_data in hint_pages:	1✔
203	# Skip high-block pages for hints too (same threshold)
204	if len(page_data.blocks) > MAX_BLOCKS_PER_PAGE:	1✔
205	continue	1✔
206
207	# TODO We are re-filtering duplicates here; optimize by changing the API
208	# to accept one list of PageData, and seperate by page_numbers.
209	kept_blocks = page_data.blocks	1✔
210	kept_blocks, _ = filter_overlapping_text_blocks(kept_blocks)	1✔
211	kept_blocks, _ = filter_duplicate_blocks(kept_blocks)	1✔
212
213	hint_pages_without_duplicates.append(	1✔
214	PageData(
215	page_number=page_data.page_number,
216	bbox=page_data.bbox,
217	blocks=kept_blocks,
218	)
219	)
220
221	# Build pages without duplicates for classification
222	pages_without_duplicates = []	1✔
223	for page_data, removed_mapping in zip(pages, removed_blocks_per_page, strict=True):	1✔
224	# We need to filter blocks that were removed by ANY filter
225	non_removed_blocks = [	1✔
226	block for block in page_data.blocks if block not in removed_mapping
227	]
228	pages_without_duplicates.append(	1✔
229	PageData(
230	page_number=page_data.page_number,
231	bbox=page_data.bbox,
232	blocks=non_removed_blocks,
233	)
234	)
235
236	# Generate hints from hint pages, histogram from pages to classify
237	font_size_hints = FontSizeHints.from_pages(hint_pages_without_duplicates)	1✔
238	page_hints = PageHintCollection.from_pages(hint_pages_without_duplicates)	1✔
239	histogram = TextHistogram.from_pages(pages_without_duplicates)	1✔
240
241	# Phase 3: Classify using the hints (on pages without duplicates)
242	config = ClassifierConfig(font_size_hints=font_size_hints, page_hints=page_hints)	1✔
243	classifier = Classifier(config)	1✔
244
245	results = []	1✔
246	for page_data, page_without_duplicates, removed_mapping in zip(	1✔
247	pages, pages_without_duplicates, removed_blocks_per_page, strict=True
248	):
249	# Handle skipped pages
250	if page_data.page_number in skipped_pages:	1✔
251	result = ClassificationResult(	1✔
252	page_data=page_data,
253	skipped_reason=(
254	f"Page has {len(page_data.blocks)} blocks, which exceeds "
255	f"the threshold of {MAX_BLOCKS_PER_PAGE}. This is likely an "
256	f"info/inventory page with vectorized text."
257	),
258	)
259	results.append(result)	1✔
260	continue	1✔
261
262	# Classify using only non-removed blocks
263	result = classifier.classify(page_without_duplicates)	1✔
264
265	# Update result to use original page_data (with all blocks)
266	result.page_data = page_data	1✔
267
268	# Mark removed blocks
269	for removed_block, removal_reason in removed_mapping.items():	1✔
270	result.mark_removed(removed_block, removal_reason)	1✔
271
272	results.append(result)	1✔
273
274	return BatchClassificationResult(results=results, histogram=histogram)	1✔
275
276
277	type Classifiers = (	1✔
278	PageNumberClassifier
279	\| ProgressBarClassifier
280	\| ProgressBarIndicatorClassifier
281	\| PreviewClassifier
282	\| BackgroundClassifier
283	\| DividerClassifier
284	\| BagNumberClassifier
285	\| PartCountClassifier
286	\| PartNumberClassifier
287	\| StepNumberClassifier
288	\| StepCountClassifier
289	\| PieceLengthClassifier
290	\| ScaleClassifier
291	\| PartsClassifier
292	\| PartsListClassifier
293	\| PartsImageClassifier
294	\| ShineClassifier
295	\| OpenBagClassifier
296	\| LoosePartSymbolClassifier
297	\| DiagramClassifier
298	\| ArrowClassifier
299	\| SubAssemblyClassifier
300	\| StepClassifier
301	\| TriviaTextClassifier
302	\| PageClassifier
303	)
304
305
306	class Classifier:	1✔
307	"""
308	Performs a single run of classification based on rules, configuration, and hints.
309	This class should be stateless.
310	"""
311
312	def __init__(self, config: ClassifierConfig):	1✔
313	self.config = config	1✔
314	# Sort classifiers topologically based on their dependencies
315	self.classifiers = topological_sort(	1✔
316	[
317	PageNumberClassifier(config=config),
318	ProgressBarIndicatorClassifier(config=config),
319	ProgressBarClassifier(config=config),
320	BackgroundClassifier(config=config),
321	DividerClassifier(config=config),
322	BagNumberClassifier(config=config),
323	PartCountClassifier(config=config),
324	PartNumberClassifier(config=config),
325	StepNumberClassifier(config=config),
326	SubStepNumberClassifier(config=config),
327	StepCountClassifier(config=config),
328	PieceLengthClassifier(config=config),
329	ScaleClassifier(config=config),
330	PartsClassifier(config=config),
331	PartsListClassifier(config=config),
332	DiagramClassifier(config=config),
333	RotationSymbolClassifier(config=config),
334	ArrowClassifier(config=config),
335	PartsImageClassifier(config=config),
336	ShineClassifier(config=config),
337	OpenBagClassifier(config=config),
338	LoosePartSymbolClassifier(config=config),
339	PreviewClassifier(config=config),
340	SubStepClassifier(config=config),
341	SubAssemblyClassifier(config=config),
342	StepClassifier(config=config),
343	TriviaTextClassifier(config=config),
344	PageClassifier(config=config),
345	]
346	)
347
348	def classify(self, page_data: PageData) -> ClassificationResult:	1✔
349	"""
350	Runs the classification logic and returns a result.
351	It does NOT modify page_data directly.
352
353	The classification process runs in three phases:
354	1. Score all classifiers (bottom-up) - auto-registers classifiers
355	2. Construct final elements (top-down starting from Page)
356	"""
357	result = ClassificationResult(page_data=page_data)	1✔
358
359	logger.debug(f"Starting classification for page {page_data.page_number}")	1✔
360
361	# 1. Score all classifiers (Bottom-Up)
362	# Note: score() automatically registers each classifier for its output labels
363	for classifier in self.classifiers:	1✔
364	classifier.score(result)	1✔
365
366	# 2. Construct (Top-Down)
367	# Find the PageClassifier to start the construction process
368	page_classifier = next(	1✔
369	c for c in self.classifiers if isinstance(c, PageClassifier)
370	)
371	page_classifier.build_all(result)	1✔
372
373	# 3. Validate that all page elements are tracked via candidates
374	# This catches programming errors where elements are created directly
375	# instead of via result.build()
376	from build_a_long.pdf_extract.validation.rules import (	1✔
377	assert_page_elements_tracked,
378	)
379
380	assert_page_elements_tracked(result)	1✔
381
382	# TODO Do we actualy ever add warnings?
383	warnings = self._log_post_classification_warnings(page_data, result)	1✔
384	for warning in warnings:	1✔
385	result.add_warning(warning)	1✔
386
387	return result	1✔
388
389	def _log_post_classification_warnings(	1✔
390	self, page_data: PageData, result: ClassificationResult
391	) -> list[str]:
392	warnings = []	1✔
393
394	# Check if there's a page number
395	page_numbers = result.get_winners_by_score("page_number", PageNumber)	1✔
396	if not page_numbers:	1✔
397	warnings.append(f"Page {page_data.page_number}: missing page number")	1✔
398
399	# Get elements by label
400	parts_lists = result.get_winners_by_score("parts_list", PartsList)	1✔
401	part_counts = result.get_winners_by_score("part_count", PartCount)	1✔
402
403	for pl in parts_lists:	1✔
404	inside_counts = filter_contained(part_counts, pl.bbox)	1✔
405	if not inside_counts:	1✔
406	warnings.append(	×
407	f"Page {page_data.page_number}: parts list at {pl.bbox} "
408	f"contains no part counts"
409	)
410
411	steps = result.get_winners_by_score("step_number", StepNumber)	1✔
412	ABOVE_EPS = 2.0	1✔
413	for step in steps:	1✔
414	sb = step.bbox	1✔
415	above = [pl for pl in parts_lists if pl.bbox.y1 <= sb.y0 + ABOVE_EPS]	1✔
416	if not above:	1✔
417	warnings.append(	1✔
418	f"Page {page_data.page_number}: step number '{step.value}' "
419	f"at {sb} has no parts list above it"
420	)
421	return warnings	1✔

bramp / build-along / 20258147097

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous