Automated Field Mapping & Data Normalization

Every Driver Vehicle Inspection Report (DVIR) that reaches the compliance store must present the same field names, the same types, and the same controlled vocabularies regardless of whether it originated on a native mobile app, a scanned paper form, or a telematics feed — because 49 CFR § 396.11(a) attaches recordkeeping obligations to specific fields (driver identity, VIN, preparation timestamp, per-component condition, and the certification signature) that downstream logic cannot enforce if a vendor labels them unitNo, unit_number, or Truck # interchangeably. Field mapping and normalization is the stage inside the DVIR Ingestion & Digital/Paper Parsing Workflows pipeline that folds every vendor-specific payload into one canonical record. When it is done loosely — coercing a missing timestamp to “now”, silently dropping an unrecognized defect code — the carrier loses the ability to prove a § 396.11© certification chain, and a mislabeled safety defect can slip past the Critical vs Non-Critical Routing Logic gate that decides whether a truck is legally allowed to move.

The mapper is a deterministic transform that runs immediately after channel decoding and before the validation gate. It does not fetch, guess, or enrich from external services during a single record’s transformation — every mapping decision must be reproducible from the payload plus a versioned rule set. Target Python 3.10+ (the code uses match statements, StrEnum, and structural typing) with:

• Pydantic 2.x — the canonical output contract and runtime validation of the normalized record.
• rapidfuzz — fuzzy alias resolution and OCR artifact correction against the controlled vocabularies.
• python-dateutil / zoneinfo (stdlib) — timezone-aware parsing of driver-entered and vendor timestamps.
• polars — high-throughput batch normalization when end-of-shift bursts are drained from the queue.
• PyYAML — externalized mapping tables so compliance officers can revise aliases and thresholds without a code deploy.

The normalized output is not a free-form dict: it must satisfy the canonical extracted-field contract owned by the Standardized DVIR JSON Schema Design reference. This page adds only the alias-resolution, cleaning, and coercion logic that gets a raw payload to that contract; defect codes themselves resolve against the controlled taxonomy in Defect Taxonomy Mapping for Heavy Trucks.

The mapper operates on two records: the loosely-typed inbound payload it consumes (an arbitrary key/value envelope from one channel) and the canonical record it emits. The table below is the canonical output contract this layer targets; field names and enumerations are held identical to the schema and defect-classification pages so a single record can be reconstructed across the pipeline during a DOT audit.

Normalization must be a pure function: identical inputs yield identical outputs with no side effects. Each raw field passes through three sequential stages before it is admitted to a canonical column.

The raw value is never discarded. Every record carries raw_payload_sha256 and the original field is preserved in the audit envelope, so a normalization decision can be reproduced and challenged during an audit.

The mapper resolves vendor keys against an externalized alias table, cleans and coerces each value, and constructs a Pydantic record that either validates or is rejected. Fold no clocks, network calls, or randomness into the transform — resolve those before or after so the mapping is replayable from the audit log.

from __future__ import annotations

import re
from datetime import datetime
from enum import StrEnum
from zoneinfo import ZoneInfo

from dateutil import parser as dateparser
from pydantic import BaseModel, Field, ValidationError, field_validator
from rapidfuzz import process, fuzz


class DefectSeverity(StrEnum):
    MINOR = "minor"
    MAJOR = "major"
    CRITICAL = "critical"


# Externalized in YAML in production; inlined here for clarity. The mapper
# NEVER hard-codes vendor keys in Python — compliance officers own this table.
FIELD_ALIASES: dict[str, tuple[str, ...]] = {
    "asset_id": ("vin", "unit", "unit_no", "unitNumber", "truck", "vehicle_id"),
    "odometer": ("odo", "mileage", "miles", "hubometer"),
    "prepared_at": ("timestamp", "inspection_time", "completed_at", "date"),
    "driver_id": ("driver", "operator", "emp_id", "driver_code"),
}

SEVERITY_VOCAB: dict[str, DefectSeverity] = {
    "s": DefectSeverity.MINOR, "minor": DefectSeverity.MINOR, "cosmetic": DefectSeverity.MINOR,
    "m": DefectSeverity.MAJOR, "major": DefectSeverity.MAJOR, "needs repair": DefectSeverity.MAJOR,
    "c": DefectSeverity.CRITICAL, "critical": DefectSeverity.CRITICAL, "oos": DefectSeverity.CRITICAL,
}


def resolve_key(raw_key: str) -> str | None:
    """Map a vendor key to its canonical field, tolerating case and punctuation."""
    needle = raw_key.strip().lower().replace(" ", "_")
    for canonical, aliases in FIELD_ALIASES.items():
        if needle == canonical or needle in {a.lower() for a in aliases}:
            return canonical
    return None


def clean_odometer(raw: str) -> str:
    """Lexical stage: drop thousands separators and trailing unit tokens."""
    return re.sub(r"[^\d]", "", raw.split(".")[0])  # "142,305 mi" -> "142305"


def map_severity(raw: str) -> DefectSeverity:
    """Semantic stage: resolve driver shorthand to the controlled enum, with a
    fuzzy fallback so 'crtical' (an OCR/typo miss) still lands on CRITICAL."""
    token = raw.strip().lower()
    if token in SEVERITY_VOCAB:
        return SEVERITY_VOCAB[token]
    match = process.extractOne(token, SEVERITY_VOCAB.keys(), scorer=fuzz.ratio, score_cutoff=80)
    if match is None:
        raise ValueError(f"unresolved severity token: {raw!r}")  # never default silently
    return SEVERITY_VOCAB[match[0]]

The canonical record is a frozen Pydantic model. Coercion that cannot validate — a VIN of the wrong length, an odometer outside the plausible range, an absent signature — raises rather than admitting a guess.

class CanonicalDVIR(BaseModel, frozen=True):
    dvir_id: str
    driver_id: str
    asset_id: str = Field(min_length=11, max_length=17)
    prepared_at: datetime
    odometer: int = Field(ge=0, le=3_000_000)
    defect_severity: DefectSeverity
    signature_present: bool
    source_channel: str
    raw_payload_sha256: str

    @field_validator("prepared_at")
    @classmethod
    def must_be_tz_aware(cls, v: datetime) -> datetime:
        # § 396.11(a) completion timestamp must be unambiguous in UTC.
        if v.tzinfo is None:
            raise ValueError("prepared_at must be timezone-aware")
        return v

    @field_validator("signature_present")
    @classmethod
    def certification_required(cls, v: bool) -> bool:
        if not v:
            raise ValueError("missing § 396.11(a) certification signature")
        return v


def normalize(raw: dict, *, default_tz: str = "UTC") -> CanonicalDVIR:
    """Pure transform: raw channel payload -> validated canonical record."""
    mapped: dict[str, object] = {}
    for k, v in raw.items():
        canonical = resolve_key(k)
        if canonical is None:
            continue  # unknown keys are preserved in the raw envelope, not the record
        match canonical:
            case "odometer":
                mapped["odometer"] = int(clean_odometer(str(v)))
            case "prepared_at":
                dt = dateparser.parse(str(v))
                mapped["prepared_at"] = dt.replace(tzinfo=ZoneInfo(default_tz)) if dt.tzinfo is None else dt
            case _:
                mapped[canonical] = v
    return CanonicalDVIR(**mapped)  # raises ValidationError if any contract is violated

When a payload arrives pre-structured from a driver-facing app, the Mobile App DVIR Export Integration usually delivers near-canonical keys, so resolve_key is a near-identity pass and only timezone harmonization is required. Scanned and handwritten submissions are the hard case: the PDF & Image OCR Pipeline Setup emits raw text plus per-token confidence, and the fuzzy fallback in map_severity is where OCR artifacts like crtical or a transposed VIN character are reconciled — or, below the confidence floor, escalated to human review rather than admitted as a guess. Free-text defect descriptions and driver shorthand that need context-aware handling are covered in depth by Normalizing Inconsistent Driver Input Fields.

Normalization is a compliance gate, not a formatting convenience: each outcome maps to a concrete FMCSA obligation, and the fail-closed default is rejection, never admission. The table below is the authoritative mapping compliance officers should reference when reconciling the dead-letter queue against inspection records.

The governing imperative: when a value is ambiguous, unresolved, or below the confidence floor, reject the payload and preserve it for correction rather than coercing it into a canonical column. A false positive here — a mislabeled critical defect normalized down to minor — is exactly what the routing gate cannot recover from.

The normalized record is a fact other systems subscribe to, not a call the mapper blocks on. Emit each CanonicalDVIR to the broker as a single event keyed by a deterministic identifier so retries during cellular dropouts or broker redelivery do not create duplicate compliance records:

import hashlib


def idempotency_key(record: CanonicalDVIR) -> str:
    # Deterministic across retries: same inspection event => same key.
    material = f"{record.dvir_id}:{record.raw_payload_sha256}"
    return hashlib.sha256(material.encode()).hexdigest()

High-volume end-of-shift bursts are drained through the machinery described in Async Batching for High-Volume Ingestion, where polars normalizes a frame of records in one vectorized pass rather than row-by-row. Downstream, the Computerized Maintenance Management System (CMMS) consumes normalized defect records to open work orders, ELD and telematics platforms (Samsara, Geotab) consume asset_id and prepared_at to reconcile inspection state against the vehicle, and the audit store consumes every record. Chain audit entries cryptographically — each stored record includes the SHA-256 hash of the previous record for the same asset — so a deleted or reordered inspection is detectable during a DOT audit. Reconciliation jobs periodically diff the normalized ledger against the raw envelope store to catch any field that was dropped or coerced without a preserved original.

• Schema validation — every emitted record satisfies the canonical Pydantic contract; a record that cannot validate is rejected, never partially admitted.
• Deterministic execution — mapping is a pure function of the payload plus a versioned rule set; no clocks, network calls, or randomness inside normalize().
• Externalized rules — alias tables and controlled vocabularies live in YAML that compliance officers revise without a code deploy; the mapper pins a rule_version.
• Raw preservation — the original bytes and their SHA-256 hash are retained alongside every normalized record for audit reproduction.
• Fail-closed defaults — ambiguous, unresolved, or low-confidence values are rejected and preserved for correction, never coerced.
• Idempotent emission — every published record carries a deterministic key derived from stable fields.
• Cryptographic chaining — hash-link consecutive records per asset for a tamper-evident audit package.

• Normalizing Inconsistent Driver Input Fields — context-aware handling of odometer, severity, and free-text driver shorthand.
• PDF & Image OCR Pipeline Setup — the source of raw text and confidence scores this layer reconciles.
• Mobile App DVIR Export Integration — delivers near-canonical payloads that need only light alias resolution.
• Async Batching for High-Volume Ingestion — vectorized batch normalization for end-of-shift bursts.
• Standardized DVIR JSON Schema Design — the canonical contract every normalized record must satisfy.

Back to DVIR Ingestion & Digital/Paper Parsing Workflows.