2. Ingesting Bharat NCAP Assessment Reports into a Structured Store

Why ingestion is a separate concern from scoring

The scoring data model from the previous step defines what a valid Bharat NCAP assessment record looks like. Ingestion answers a different question: given the raw assessment reports that testing agencies publish — typically PDFs or semi-structured exports containing per-test point tables — how do we reliably produce records that satisfy that model? The two concerns must stay separate. If parsing logic leaks into the scoring layer, every new report format forces changes to star-rating computation. If validation is skipped during ingestion, malformed rows silently poison downstream aggregation and produce plausible-looking but wrong star ratings.

This case builds a pipeline with three explicit stages: extract (pull tabular test data out of a report), normalize (map heterogeneous test types onto the canonical model), and validate-and-store (reject incomplete records, persist the rest with their per-test point breakdowns intact). The critical requirement is that per-test point breakdowns survive ingestion. A vehicle's Adult Occupant Protection score is not a single number in the source report — it is the sum of points earned across frontal offset, side impact, and pole tests, each with its own sub-items. The dashboard later needs those sub-items to explain why a score is what it is, so flattening them at ingestion time is a defect, not a simplification.

The canonical record shape

Assume the scoring model exposes a record with a vehicle identity block, a category block, and a list of test results. Each test result carries a test type, the maximum available points, the points awarded, and a breakdown of sub-items. The ingestion pipeline's job is to emit exactly this shape. A minimal but complete representation in Python:

from dataclasses import dataclass, field
from typing import Literal

TestType = Literal["frontal_offset", "side_impact", "side_pole", "pedestrian", "esc", "seatbelt_reminder"]

@dataclass
class SubItem:
    name: str
    max_points: float
    awarded: float

@dataclass
class TestResult:
    test_type: TestType
    max_points: float
    awarded: float
    sub_items: list[SubItem] = field(default_factory=list)

@dataclass
class AssessmentRecord:
    vehicle_id: str
    make: str
    model: str
    variant: str
    category: Literal["AOP", "COP", "SA"]
    tests: list[TestResult] = field(default_factory=list)

    def category_score(self) -> float:
        return sum(t.awarded for t in self.tests)

Note that category_score() is derived, never stored. Storing a computed total alongside its components invites drift: a corrected sub-item updates the breakdown but not the cached total. Derive at read time.

Extraction: handling multiple test types

Real assessment reports mix test types within one category. An AOP section might contain a frontal offset test scored out of 16, a side impact test out of 16, and a pole test out of 2, each with sub-items like "head", "chest", "legs". The extractor must not assume a fixed column layout. A robust approach is to parse each report into a list of raw rows keyed by a detected test-type header, then hand those rows to a per-type normalizer.

import re

TEST_HEADERS = {
    "frontal offset": "frontal_offset",
    "side impact": "side_impact",
    "side pole": "side_pole",
    "pedestrian protection": "pedestrian",
    "electronic stability control": "esc",
    "seat belt reminder": "seatbelt_reminder",
}

def detect_test_type(line: str) -> str | None:
    low = line.lower()
    for phrase, canonical in TEST_HEADERS.items():
        if phrase in low:
            return canonical
    return None

def split_into_sections(raw_lines: list[str]) -> dict[str, list[str]]:
    sections: dict[str, list[str]] = {}
    current = None
    for line in raw_lines:
        detected = detect_test_type(line)
        if detected:
            current = detected
            sections.setdefault(current, [])
        elif current and line.strip():
            sections[current].append(line.strip())
    return sections

The detect_test_type function is deliberately tolerant: it matches on phrases rather than exact strings, because report headings vary ("Frontal Offset Deformable Barrier" vs "Frontal Offset Test"). Unknown headings are skipped rather than guessed — an unrecognized section is a signal to extend the header map, not to coerce it into the nearest known type.

Normalization and the point-breakdown contract

Each section's rows are parsed into SubItem objects. The contract is strict: the sum of sub-item max_points must equal the test's declared maximum, and the sum of awarded must equal the test's awarded total. If either invariant fails, the record is rejected with a diagnostic naming the offending test. This catches the most common extraction failure — a sub-item row silently dropped because its label wrapped onto two lines.

def build_test_result(test_type: str, rows: list[str]) -> TestResult:
    sub_items = []
    for row in rows:
        m = re.match(r"(.+?)\s+([\d.]+)\s*/\s*([\d.]+)", row)
        if not m:
            continue
        name, awarded, max_pts = m.group(1).strip(), float(m.group(2)), float(m.group(3))
        sub_items.append(SubItem(name=name, max_points=max_pts, awarded=awarded))
    if not sub_items:
        raise ValueError(f"no sub-items parsed for {test_type}")
    total_max = sum(s.max_points for s in sub_items)
    total_awarded = sum(s.awarded for s in sub_items)
    return TestResult(test_type=test_type, max_points=total_max,
                      awarded=total_awarded, sub_items=sub_items)

The regex expects the common "label 12.5 / 16" layout. Reports that use separate columns require a different row parser, but the invariant check downstream is format-agnostic — that is the point of separating extraction from validation.

Validation and storage

Validation runs after normalization and before persistence. It checks two things the extractor cannot: that every test's sub-item totals reconcile, and that the assembled AssessmentRecord satisfies the schema from the scoring model (required fields present, category in the allowed set, no duplicate test types within a category). Records that fail are written to a diagnostics log with the vehicle ID, the failing test type, and the observed versus expected totals. They are not partially stored — a half-ingested vehicle is worse than an absent one, because the dashboard would render it with a misleadingly low score.

def ingest(raw_lines: list[str], identity: dict) -> AssessmentRecord:
    sections = split_into_sections(raw_lines)
    tests = []
    for test_type, rows in sections.items():
        result = build_test_result(test_type, rows)
        if abs(result.max_points - result.awarded) > 0 and result.awarded > result.max_points:
            raise ValueError(f"{test_type}: awarded {result.awarded} exceeds max {result.max_points}")
        tests.append(result)
    record = AssessmentRecord(**identity, tests=tests)
    seen = set()
    for t in record.tests:
        if t.test_type in seen:
            raise ValueError(f"duplicate test type {t.test_type}")
        seen.add(t.test_type)
    return record

The identity dict supplies vehicle_id, make, model, variant, and category from the report's cover metadata, which is parsed separately from the test tables. Keeping identity extraction out of the section parser avoids coupling the two when report layouts change independently.

Verifying the pipeline

A reproducible check uses a small synthetic report with two AOP tests: a frontal offset with sub-items 8/8, 4/4, 2/4 (total 14/16) and a side impact with 10/10, 4/6 (total 14/16). After ingestion, assert that record.category_score() == 28.0, that len(record.tests) == 2, and that the frontal offset test retains three sub-items. Then corrupt one row by deleting the "2 / 4" line and confirm ingestion raises with a message naming frontal_offset — the invariant catches the dropped row because the sub-item sum no longer matches the declared maximum. These assertions are the acceptance criteria: a pipeline that passes them preserves breakdowns and rejects incomplete data, which is exactly what downstream aggregation depends on.