20472199734

Committed 23 Dec 2025 09:39PM UTC coverage: 88.693% (+0.2%) from 88.542%

Build # 20472199734

Build Type

push

github

Committed by

bramp

Commit Message

Add no-orphan constraints for Step child elements

- Add no-orphan constraints to StepClassifier.declare_constraints() for:
  - arrows (point from subassembly callouts to main diagram)
  - rotation_symbols (indicate model rotation)
  - subassemblies (callout boxes within steps)
  - substeps (mini-steps within a main step)
  - diagrams (the main instruction graphic)

- If any of these elements are selected, at least one step must also be
  selected, preventing orphaned elements

- Add unit tests for no-orphan constraint declaration

- Update architecture docs with no-orphan constraint documentation

- Add TODO for potential future centralization in SchemaConstraintGenerator

Run Details

66 of 67 new or added lines in 2 files covered. (98.51%)

151 existing lines in 8 files now uncovered.

14786 of 16671 relevant lines covered (88.69%)

0.89 hits per line

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

98.82

/src/build_a_long/pdf_extract/classifier/steps/step_classifier_test.py

from collections.abc import Callable

import pytest

from build_a_long.pdf_extract.classifier import (
    ClassificationResult,
    ClassifierConfig,
)
from build_a_long.pdf_extract.classifier.candidate import Candidate
from build_a_long.pdf_extract.classifier.conftest import CandidateFactory
from build_a_long.pdf_extract.classifier.constraint_model import ConstraintModel
from build_a_long.pdf_extract.classifier.schema_constraint_generator import (
    SchemaConstraintGenerator,
)
from build_a_long.pdf_extract.classifier.score import Score
from build_a_long.pdf_extract.classifier.steps.arrow_classifier import (
    ArrowClassifier,
    _ArrowHeadData,
    _ArrowScore,
)
from build_a_long.pdf_extract.classifier.steps.step_classifier import (
    StepClassifier,
    _StepScore,
    filter_subassembly_values,
)
from build_a_long.pdf_extract.extractor import PageData
from build_a_long.pdf_extract.extractor.bbox import BBox
from build_a_long.pdf_extract.extractor.lego_page_elements import (
    Step,
)
from build_a_long.pdf_extract.extractor.page_blocks import Drawing, Image, Text


def run_solver(
    result: ClassificationResult,
    classifier: StepClassifier,
    labels_to_solve: set[str],
) -> None:
    """Run the constraint solver on the classification result.

    This helper mimics what Classifier._solve_constraints() does, allowing
    unit tests to test solver behavior without going through the full
    classification pipeline.

    Args:
        result: The classification result with scored candidates
        classifier: The step classifier to declare constraints
        labels_to_solve: Set of labels to include in the solver
    """
    model = ConstraintModel()

    # Add all candidates for the specified labels
    all_candidates: list[Candidate] = []
    for label, candidates in result.candidates.items():
        if label in labels_to_solve:
            for candidate in candidates:
                model.add_candidate(candidate)
                all_candidates.append(candidate)

    # Add block exclusivity constraints
    model.add_block_exclusivity_constraints(all_candidates)

    # Let the classifier declare custom constraints
    classifier.declare_constraints(model, result)

    # Auto-generate schema-based constraints
    generator = SchemaConstraintGenerator()
    generator.generate_for_classifier(classifier, model, result)

    # Add child uniqueness constraints (each child has at most one parent)
    generator.add_child_uniqueness_constraints(model)

    # Maximize total score
    model.maximize([(cand, int(cand.score * 1000)) for cand in all_candidates])

    # Solve
    solved, selection = model.solve()

    if not solved:
        result.set_solver_selection([], labels_to_solve)
        return

    # Mark selected candidates
    selected_candidates = [
        cand for cand in all_candidates if selection.get(cand.id, False)
    ]
    result.set_solver_selection(selected_candidates, labels_to_solve)


def score_solve_and_build(
    result: ClassificationResult,
    classifier: StepClassifier,
    labels_to_solve: set[str] | None = None,
) -> None:
    """Score candidates, run the solver, and build all steps.

    This is a convenience helper that combines the three-step process:
    1. classifier.score(result) - Score all step candidates
    2. run_solver() - Run constraint solver to select best candidates
    3. classifier.build_all() - Build the selected step elements

    Args:
        result: The classification result with dependency candidates already added
        classifier: The step classifier to use
        labels_to_solve: Set of labels to include in solver. Defaults to {"step"}.
    """
    if labels_to_solve is None:
        labels_to_solve = {"step"}

    classifier.score(result)
    run_solver(result, classifier, labels_to_solve)
    classifier.build_all(result)


@pytest.fixture
def classifier() -> StepClassifier:
    return StepClassifier(config=ClassifierConfig())


class TestStepClassification:
    """Tests for detecting complete Step structures."""

    def test_step_with_parts_list(
        self,
        classifier: StepClassifier,
        candidate_factory: Callable[[ClassificationResult], CandidateFactory],
    ) -> None:
        """Test a step that has an associated parts list."""
        page_bbox = BBox(0, 0, 200, 300)
        step_text = Text(id=1, bbox=BBox(50, 180, 70, 210), text="10")
        d1 = Drawing(id=2, bbox=BBox(30, 100, 170, 160))
        pc1_text = Text(id=3, bbox=BBox(40, 110, 55, 120), text="2x")
        pc2_text = Text(id=4, bbox=BBox(100, 130, 115, 140), text="5x")
        img1 = Image(id=5, bbox=BBox(40, 90, 55, 105))
        img2 = Image(id=6, bbox=BBox(100, 115, 115, 125))

        page_data = PageData(
            page_number=6,
            blocks=[step_text, d1, pc1_text, pc2_text, img1, img2],
            bbox=page_bbox,
        )

        result = ClassificationResult(page_data=page_data)
        # Register all relevant classifiers

        factory = candidate_factory(result)

        # Manually score dependencies
        factory.add_step_number(step_text)

        pc1_candidate = factory.add_part_count(pc1_text)
        pc2_candidate = factory.add_part_count(pc2_text)

        part1_candidate = factory.add_part(pc1_candidate, img1)
        part2_candidate = factory.add_part(pc2_candidate, img2)

        factory.add_parts_list(d1, [part1_candidate, part2_candidate])

        score_solve_and_build(result, classifier)

        # Get constructed steps
        steps = [
            result.get_constructed(c)
            for c in result.get_candidates("step")
            if isinstance(result.get_constructed(c), Step)
        ]
        assert len(steps) == 1
        constructed_step = steps[0]
        assert isinstance(constructed_step, Step)

        assert constructed_step.step_number.value == 10
        assert constructed_step.parts_list is not None
        assert len(constructed_step.parts_list.parts) == 2
        # Diagram is None when DiagramClassifier doesn't find any diagrams
        assert constructed_step.diagram is None

    def test_step_without_parts_list(
        self,
        classifier: StepClassifier,
        candidate_factory: Callable[[ClassificationResult], CandidateFactory],
    ) -> None:
        """Test a step that has no associated parts list."""
        page_bbox = BBox(0, 0, 200, 300)
        step_text = Text(id=1, bbox=BBox(50, 180, 70, 210), text="5")

        page_data = PageData(
            page_number=6,
            blocks=[step_text],
            bbox=page_bbox,
        )

        result = ClassificationResult(page_data=page_data)
        # Register all relevant classifiers

        factory = candidate_factory(result)

        # Manually score step number candidate
        factory.add_step_number(step_text)

        score_solve_and_build(result, classifier)

        # Get constructed steps
        steps = [
            result.get_constructed(c)
            for c in result.get_candidates("step")
            if isinstance(result.get_constructed(c), Step)
        ]
        assert len(steps) == 1
        constructed_step = steps[0]
        assert isinstance(constructed_step, Step)

        assert constructed_step.step_number.value == 5
        assert constructed_step.parts_list is None  # No parts list candidate
        # Diagram is None when DiagramClassifier doesn't find any diagrams
        assert constructed_step.diagram is None

    def test_multiple_steps_on_page(
        self,
        classifier: StepClassifier,
        candidate_factory: Callable[[ClassificationResult], CandidateFactory],
    ) -> None:
        """Test a page with multiple steps."""
        page_bbox = BBox(0, 0, 400, 300)

        # First step components
        step1_text = Text(id=1, bbox=BBox(50, 180, 70, 210), text="1")
        d1 = Drawing(id=2, bbox=BBox(30, 100, 170, 160))
        pc1_text = Text(id=3, bbox=BBox(40, 110, 55, 120), text="2x")
        img1 = Image(id=7, bbox=BBox(40, 90, 55, 105))

        # Second step components
        step2_text = Text(id=4, bbox=BBox(250, 180, 270, 210), text="2")
        d2 = Drawing(id=5, bbox=BBox(230, 100, 370, 160))
        pc2_text = Text(id=6, bbox=BBox(240, 110, 255, 120), text="3x")
        img2 = Image(id=8, bbox=BBox(240, 90, 255, 105))

        page_data = PageData(
            page_number=6,
            blocks=[step1_text, d1, pc1_text, img1, step2_text, d2, pc2_text, img2],
            bbox=page_bbox,
        )

        result = ClassificationResult(page_data=page_data)
        # Register all relevant classifiers

        factory = candidate_factory(result)

        # Score dependencies
        factory.add_step_number(step1_text)
        factory.add_step_number(step2_text)

        pc1_candidate = factory.add_part_count(pc1_text)
        pc2_candidate = factory.add_part_count(pc2_text)

        part1_candidate = factory.add_part(pc1_candidate, img1)
        part2_candidate = factory.add_part(pc2_candidate, img2)

        factory.add_parts_list(d1, [part1_candidate])
        factory.add_parts_list(d2, [part2_candidate])

        score_solve_and_build(result, classifier)

        # Get constructed steps
        steps = [
            result.get_constructed(c)
            for c in result.get_candidates("step")
            if isinstance(result.get_constructed(c), Step)
        ]
        assert len(steps) == 2

        # Check that steps are in order by value
        # Filter to Steps only for type safety
        step_objs = [s for s in steps if isinstance(s, Step)]
        steps_sorted = sorted(step_objs, key=lambda s: s.step_number.value)
        assert steps_sorted[0].step_number.value == 1
        assert steps_sorted[1].step_number.value == 2

    def test_step_score_ordering(
        self,
        classifier: StepClassifier,
        candidate_factory: Callable[[ClassificationResult], CandidateFactory],
    ) -> None:
        """Test that steps are ordered correctly by their score."""
        page_bbox = BBox(0, 0, 400, 300)

        # Step 2 appears first in the element list
        step2_text = Text(id=1, bbox=BBox(250, 180, 270, 210), text="2")

        # Step 1 appears second
        step1_text = Text(id=2, bbox=BBox(50, 180, 70, 210), text="1")

        page_data = PageData(
            page_number=6,
            blocks=[step2_text, step1_text],
            bbox=page_bbox,
        )

        result = ClassificationResult(page_data=page_data)
        # Register all relevant classifiers

        factory = candidate_factory(result)

        # Score dependencies - assign higher score to step1
        # to ensure it wins if scores are tie-broken
        factory.add_step_number(step2_text, score=0.8)  # Lower score for step2
        factory.add_step_number(step1_text, score=1.0)  # Higher score for step1

        classifier.score(result)

        # Get the candidates in sorted order
        step_candidates = result.get_candidates("step")
        sorted_candidates = sorted(
            step_candidates,
            key=lambda c: (
                c.score_details.sort_key()
                if isinstance(c.score_details, _StepScore)
                else (0.0, 0)
            ),
        )

        # Construct both steps
        assert len(sorted_candidates) >= 2
        constructed_step1 = result.build(sorted_candidates[0])
        constructed_step2 = result.build(sorted_candidates[1])

        assert isinstance(constructed_step1, Step)
        assert isinstance(constructed_step2, Step)
        assert constructed_step1.step_number.value == 1
        assert constructed_step2.step_number.value == 2

    def test_duplicate_step_numbers_only_match_one_step(
        self,
        classifier: StepClassifier,
        candidate_factory: Callable[[ClassificationResult], CandidateFactory],
    ) -> None:
        """When there are duplicate step numbers (same value), only one Step
        should be created. The StepClassifier should prefer the best-scoring
        StepNumber and skip subsequent ones with the same value.

        This test verifies that the uniqueness constraint is enforced at the
        Step level, not the PartsList level.
        """
        page_bbox = BBox(0, 0, 600, 400)

        # Two step numbers with the SAME value (both are "1")
        step1_text = Text(id=1, bbox=BBox(50, 150, 70, 180), text="1")
        step2_text = Text(
            id=2, bbox=BBox(50, 300, 70, 330), text="1"
        )  # Duplicate value

        # Two drawings, each above one of the step numbers
        d1 = Drawing(id=3, bbox=BBox(30, 80, 170, 140))  # Above step1
        d2 = Drawing(id=4, bbox=BBox(30, 230, 170, 290))  # Above step2

        # Part counts and images inside d1
        pc1_text = Text(id=5, bbox=BBox(40, 100, 55, 110), text="2x")
        img1 = Image(id=6, bbox=BBox(40, 85, 55, 95))

        # Part counts and images inside d2
        pc2_text = Text(id=7, bbox=BBox(40, 250, 55, 260), text="3x")
        img2 = Image(id=8, bbox=BBox(40, 235, 55, 245))

        page_data = PageData(
            page_number=1,
            blocks=[step1_text, step2_text, d1, d2, pc1_text, img1, pc2_text, img2],
            bbox=page_bbox,
        )

        result = ClassificationResult(page_data=page_data)
        # Register all relevant classifiers

        factory = candidate_factory(result)

        # Score dependencies
        factory.add_step_number(step1_text, score=1.0)  # Best scoring step number
        factory.add_step_number(step2_text, score=0.9)  # Lower scoring duplicate

        pc1_candidate = factory.add_part_count(pc1_text)
        pc2_candidate = factory.add_part_count(pc2_text)

        part1_candidate = factory.add_part(pc1_candidate, img1)
        part2_candidate = factory.add_part(pc2_candidate, img2)

        factory.add_parts_list(d1, [part1_candidate])
        factory.add_parts_list(d2, [part2_candidate])

        score_solve_and_build(result, classifier)

        # Get constructed steps
        steps = [
            result.get_constructed(c)
            for c in result.get_candidates("step")
            if isinstance(result.get_constructed(c), Step)
        ]

        # Only ONE step should be created (uniqueness enforced at Step level)
        assert len(steps) == 1
        step = steps[0]
        assert isinstance(step, Step)
        assert step.step_number.value == 1


class TestFilterSubassemblyValues:
    """Tests for filter_subassembly_values function.

    This function filters out items with values likely to be subassembly steps
    (e.g., 1, 2) when the list also contains higher-numbered page-level values
    (e.g., 15, 16).
    """

    def test_empty_list_returns_empty(self) -> None:
        """An empty list should return empty."""
        assert filter_subassembly_values([]) == []

    def test_single_item_returns_unchanged(self) -> None:
        """A single item should be returned unchanged."""
        items = [(5, "a")]
        assert filter_subassembly_values(items) == items

    def test_consecutive_values_no_gap_returns_unchanged(self) -> None:
        """Values with no significant gap (e.g., 15, 16, 17) return unchanged."""
        items = [(15, "a"), (16, "b"), (17, "c")]
        assert filter_subassembly_values(items) == items

    def test_gap_exactly_3_does_not_filter(self) -> None:
        """A gap of exactly 3 (not > 3) should not filter."""
        # Gap of 3: 1 -> 4 (4 - 1 = 3, not > 3)
        items = [(1, "a"), (4, "b")]
        assert filter_subassembly_values(items) == items

    def test_gap_exactly_4_filters(self) -> None:
        """A gap of exactly 4 (> 3) should filter when min_value <= 3."""
        # Gap of 4: 1 -> 5 (5 - 1 = 4, which is > 3)
        items = [(1, "a"), (5, "b")]
        assert filter_subassembly_values(items) == [(5, "b")]

    def test_gap_but_min_value_greater_than_3_no_filter(self) -> None:
        """Gap > 3 but min_value > 3 should not filter (e.g., 5, 6, 15, 16)."""
        # Gap of 9 (15 - 6 = 9) but min is 5 (> 3)
        items = [(5, "a"), (6, "b"), (15, "c"), (16, "d")]
        assert filter_subassembly_values(items) == items

    def test_min_value_exactly_3_filters(self) -> None:
        """Gap > 3 and min_value == 3 should filter (3 <= 3)."""
        # Gap of 12 (15 - 3 = 12) and min is 3 (<= 3)
        items = [(3, "a"), (15, "b"), (16, "c")]
        result = filter_subassembly_values(items)
        assert result == [(15, "b"), (16, "c")]

    def test_min_value_exactly_4_no_filter(self) -> None:
        """Gap > 3 but min_value == 4 should NOT filter (4 > 3)."""
        # Gap of 11 (15 - 4 = 11) but min is 4 (> 3)
        items = [(4, "a"), (15, "b"), (16, "c")]
        assert filter_subassembly_values(items) == items

    def test_typical_subassembly_case(self) -> None:
        """Typical case: steps 1, 2 (subassembly) + 15, 16 (page-level)."""
        items = [(1, "a"), (2, "b"), (15, "c"), (16, "d")]
        result = filter_subassembly_values(items)
        assert result == [(15, "c"), (16, "d")]

    def test_multiple_gaps_uses_largest(self) -> None:
        """When there are multiple gaps, the largest one determines filtering."""
        # Values: 1, 2, 5, 20, 21
        # Gaps: 2->5 = 3, 5->20 = 15 (largest)
        items = [(1, "a"), (2, "b"), (5, "c"), (20, "d"), (21, "e")]
        result = filter_subassembly_values(items)
        # Largest gap is 5->20, threshold = 20
        assert result == [(20, "d"), (21, "e")]

    def test_unordered_input_handled_correctly(self) -> None:
        """Items passed in non-sorted order should be handled correctly."""
        # Pass in non-sorted order
        items = [(16, "d"), (1, "a"), (15, "c"), (2, "b")]
        result = filter_subassembly_values(items)
        # Should filter and return sorted
        assert result == [(15, "c"), (16, "d")]

    def test_preserves_associated_data(self) -> None:
        """The associated data (second element of tuple) should be preserved."""
        items = [
            (1, {"name": "step1"}),
            (2, {"name": "step2"}),
            (15, {"name": "step15"}),
        ]
        result = filter_subassembly_values(items)
        assert result == [(15, {"name": "step15"})]

    def test_custom_min_gap_parameter(self) -> None:
        """Custom min_gap parameter should be respected."""
        items = [(1, "a"), (3, "b")]  # Gap of 2
        # Default min_gap=3, so gap of 2 doesn't filter
        assert filter_subassembly_values(items) == items
        # With min_gap=1, gap of 2 > 1 should filter
        assert filter_subassembly_values(items, min_gap=1) == [(3, "b")]

    def test_custom_max_subassembly_start_parameter(self) -> None:
        """Custom max_subassembly_start parameter should be respected."""
        items = [(4, "a"), (15, "b")]  # Gap of 11, min=4
        # Default max_subassembly_start=3, so min=4 doesn't filter
        assert filter_subassembly_values(items) == items
        # With max_subassembly_start=4, min=4 should filter
        assert filter_subassembly_values(items, max_subassembly_start=4) == [(15, "b")]


class TestNoOrphanConstraints:
    """Tests for no-orphan constraints in StepClassifier."""

    def test_no_orphan_constraint_declared_for_arrows(
        self,
        classifier: StepClassifier,
        candidate_factory: Callable[[ClassificationResult], CandidateFactory],
    ) -> None:
        """Verify no-orphan constraint is declared for arrows.

        When arrows exist, the constraint solver must ensure at least one
        step is selected to prevent orphaned arrows.
        """
        page_bbox = BBox(0, 0, 600, 400)
        step_text = Text(id=1, bbox=BBox(50, 50, 70, 70), text="1")
        arrow_drawing = Drawing(id=2, bbox=BBox(100, 100, 200, 110))

        page_data = PageData(
            page_number=1,
            blocks=[step_text, arrow_drawing],
            bbox=page_bbox,
        )

        result = ClassificationResult(page_data=page_data)
        factory = candidate_factory(result)

        # Add step_number and step candidates
        factory.add_step_number(step_text)
        classifier.score(result)

        # Add an arrow candidate manually
        result._register_classifier("arrow", ArrowClassifier(config=ClassifierConfig()))
        arrow_score = _ArrowScore(
            heads=[
                _ArrowHeadData(
                    tip=(150, 105),
                    direction=0,
                    shape_score=1.0,
                    size_score=1.0,
                    block=arrow_drawing,
                )
            ]
        )
        arrow_cand = Candidate(
            bbox=arrow_drawing.bbox,
            label="arrow",
            score=1.0,
            score_details=arrow_score,
            source_blocks=[arrow_drawing],
        )
        result.add_candidate(arrow_cand)

        # Create constraint model
        model = ConstraintModel()
        for cand in result.get_candidates("step"):
            model.add_candidate(cand)
        model.add_candidate(arrow_cand)

        # Declare constraints
        classifier.declare_constraints(model, result)

        # Verify constraints were added by checking model internals
        # (if_any_selected_then_one_of creates indicator variables)
        assert model._constraint_counts.get("if_any_selected_then_one_of", 0) >= 1

    def test_no_orphan_constraints_for_all_child_elements(
        self,
        classifier: StepClassifier,
        candidate_factory: Callable[[ClassificationResult], CandidateFactory],
    ) -> None:
        """Verify no-orphan constraints are declared for all relevant element types."""
        page_bbox = BBox(0, 0, 600, 400)
        step_text = Text(id=1, bbox=BBox(50, 50, 70, 70), text="1")

        # Create dummy blocks for simple orphan-able element types
        # Note: substep is a composite element with empty source_blocks
        simple_orphan_labels = ["arrow", "rotation_symbol", "subassembly", "diagram"]
        dummy_blocks = [
            Drawing(id=100 + i, bbox=BBox(200 + i * 50, 100, 250 + i * 50, 150))
            for i in range(len(simple_orphan_labels))
        ]

        page_data = PageData(
            page_number=1,
            blocks=[step_text, *dummy_blocks],
            bbox=page_bbox,
        )

        result = ClassificationResult(page_data=page_data)
        factory = candidate_factory(result)

        # Add step_number and step candidates
        factory.add_step_number(step_text)
        classifier.score(result)

        # Create a concrete Score subclass for testing
        class _DummyScore(Score):
            def score(self) -> float:
                return 1.0

        # Add simple element candidates (with source_blocks)
        for i, label in enumerate(simple_orphan_labels):
            dummy_cand = Candidate(
                bbox=dummy_blocks[i].bbox,
                label=label,
                score=1.0,
                score_details=_DummyScore(),
                source_blocks=[dummy_blocks[i]],
            )
            result.add_candidate(dummy_cand)

        # Add substep candidate (composite, empty source_blocks)
        substep_cand = Candidate(
            bbox=BBox(300, 200, 400, 300),
            label="substep",
            score=1.0,
            score_details=_DummyScore(),
            source_blocks=[],  # Composite elements have empty source_blocks
        )
        result.add_candidate(substep_cand)

        # All orphan labels we expect constraints for
        all_orphan_labels = [
            "arrow",
            "rotation_symbol",
            "subassembly",
            "substep",
            "diagram",
        ]

        # Create constraint model
        model = ConstraintModel()
        for cand in result.get_candidates("step"):
            model.add_candidate(cand)
        for label in all_orphan_labels:
            for cand in result.get_candidates(label):
                model.add_candidate(cand)

        # Declare constraints
        classifier.declare_constraints(model, result)

        # Should have 5 no-orphan constraints (one per orphan label)
        assert model._constraint_counts.get("if_any_selected_then_one_of", 0) == 5

bramp / build-along / 20472199734

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