Defect Code Standardization Across Fleets

How do you resolve the same brake-defect concept written five different ways — FRT-014, VOLVO_BRAKE_LOW, brakes out of adjustment, BRK/ADJ, and a bare 14 — into a single canonical code that a compliance engine can act on deterministically? Get this wrong and the consequence is not cosmetic: an unmapped or mis-mapped defect that should have triggered an out-of-service (OOS) hold under 49 CFR § 396.11©(2) instead flows through as an advisory note, the tractor dispatches, and the fleet carries a documented but unactioned safety-critical defect into a roadside inspection. Silent mis-mapping is worse than rejection, because it produces an audit trail that looks compliant while the underlying decision was wrong. This page is the code-resolution implementation under the Defect Taxonomy Mapping for Heavy Trucks specification, itself part of the Core DVIR Architecture & FMCSA Compliance Mapping reference architecture.

The goal is narrow and testable: given any raw defect string from any driver, mechanic, ELD vendor, or legacy telematics export, return exactly one canonical defect code plus its severity band, or route the string to a quarantine queue for human review. No third state. Never guess into a compliance action.

Before implementing the resolver you need:

• Python 3.10+ (the code uses match/case and the X | Y union syntax).
• A canonical taxonomy already defined. The canonical codes and their severity bands come from the parent Defect Taxonomy Mapping for Heavy Trucks hierarchy — do not invent codes here, resolve into that set.
• A validated payload contract. Defect strings should already have passed structural validation via the Standardized DVIR JSON Schema Design; this resolver operates on the defects[].raw_code field after schema validation, not before.
• PyYAML for the mapping registry (pip install pyyaml) and rapidfuzz for the bounded fuzzy fallback (pip install rapidfuzz) — rapidfuzz is preferred over python-Levenshtein for its C-backed Damerau-Levenshtein and permissive licence.
• The severity bands used site-wide: 0–34 minor, 35–69 major, 70–100 critical. A critical band is the trigger for the OOS evaluation described in Critical vs Non-Critical Routing Logic.

Hardcoded translation dictionaries rot. Every new OEM firmware revision or acquired sub-fleet adds aliases, and a code change per alias is not defensible in an audit. Externalise the mapping into a YAML registry that is the single source of truth. Each entry declares one canonical code, its severity band, its OOS flag, an ordered list of literal aliases, and optional regex extraction patterns for structured vendor formats.

# defect_registry.yaml — versioned, JSON-Schema-validated at load time
version: 3
defects:
  - canonical: BRK_ADJ_OOA          # brakes out of adjustment
    severity_band: critical         # 70-100: OOS-eligible per 49 CFR 393.47
    oos_eligible: true
    aliases: ["brakes out of adjustment", "brk/adj", "brake adj", "ooa"]
    patterns: ['^FRT-0*14$', '^VOLVO_BRAKE_LOW$']
  - canonical: LMP_HEADLAMP_OUT
    severity_band: minor            # 0-34: scheduled maintenance
    oos_eligible: false
    aliases: ["headlight out", "head lamp inop", "hl out"]
    patterns: ['^LT-HDLMP-\d+$']
  - canonical: TIRE_TREAD_BELOW_MIN
    severity_band: critical         # 70-100: OOS per 49 CFR 393.75(b)
    oos_eligible: true
    aliases: ["bald tire", "tread below 2/32", "tire worn"]
    patterns: ['^TR-\d{2}-LOW$']

Validate the registry structure once, at process start, and build the fast in-memory lookup tables. A malformed alias array must fail loudly at deploy time, never at request time when it would corrupt a live compliance decision.

import re
import unicodedata
from dataclasses import dataclass
from pathlib import Path

import yaml

SEVERITY_BANDS = {"minor": (0, 34), "major": (35, 69), "critical": (70, 100)}


@dataclass(frozen=True, slots=True)
class CanonicalDefect:
    canonical: str
    severity_band: str
    oos_eligible: bool


class DefectRegistry:
    def __init__(self, path: str | Path) -> None:
        data = yaml.safe_load(Path(path).read_text(encoding="utf-8"))
        self.version: int = data["version"]
        self._by_alias: dict[str, CanonicalDefect] = {}
        self._patterns: list[tuple[re.Pattern[str], CanonicalDefect]] = []

        for entry in data["defects"]:
            band = entry["severity_band"]
            if band not in SEVERITY_BANDS:                  # fail loud at load
                raise ValueError(f"unknown severity_band {band!r} for {entry['canonical']}")
            defect = CanonicalDefect(entry["canonical"], band, bool(entry["oos_eligible"]))
            # Aliases are pre-normalized so lookups are O(1) at request time.
            for alias in entry.get("aliases", []):
                self._by_alias[normalize(alias)] = defect
            for pat in entry.get("patterns", []):
                self._patterns.append((re.compile(pat, re.IGNORECASE), defect))


def normalize(raw: str) -> str:
    """Deterministic canonical form: NFC, lowercase, collapse internal whitespace."""
    text = unicodedata.normalize("NFC", raw).strip().lower()
    return re.sub(r"\s+", " ", text)

Resolution proceeds through progressively looser strategies, and the order is the contract: an exact literal match must always win over a regex match, which must always win over a fuzzy guess. Anything that reaches the bottom of the ladder is quarantined — it is never mapped by proximity into a compliance action.

from rapidfuzz.distance import DamerauLevenshtein

FUZZY_MAX_EDITS = 2   # typo tolerance for mobile-keyboard input; tighten, never loosen


@dataclass(frozen=True, slots=True)
class Resolution:
    defect: CanonicalDefect | None   # None => quarantine
    stage: str                       # "exact" | "pattern" | "fuzzy" | "quarantine"
    confidence: float                # 1.0 exact/pattern; <1.0 fuzzy; 0.0 quarantine


def resolve(raw: str, reg: DefectRegistry) -> Resolution:
    key = normalize(raw)

    # 1. Exact alias lookup — highest confidence, O(1).
    if hit := reg._by_alias.get(key):
        return Resolution(hit, "exact", 1.0)

    # 2. Structured vendor formats via regex (e.g. FRT-014, VOLVO_BRAKE_LOW).
    for pattern, defect in reg._patterns:
        if pattern.fullmatch(raw.strip()):
            return Resolution(defect, "pattern", 1.0)

    # 3. Bounded fuzzy fallback — only for typos, never for semantic guesses.
    best, best_dist = None, FUZZY_MAX_EDITS + 1
    for alias, defect in reg._by_alias.items():
        dist = DamerauLevenshtein.distance(key, alias, score_cutoff=FUZZY_MAX_EDITS)
        if dist < best_dist:
            best, best_dist = defect, dist
    if best is not None and best_dist <= FUZZY_MAX_EDITS:
        return Resolution(best, "fuzzy", 1.0 - best_dist / (FUZZY_MAX_EDITS + 1))

    # 4. No confident match — quarantine, do NOT map into a compliance action.
    return Resolution(None, "quarantine", 0.0)

Drivers and legacy exports frequently pack several defects into one field: LIGHTS_BRAKES_TIRES or STEERING+EXHAUST. Single-match logic silently drops all but one. Split on the common delimiters, resolve each token independently, and — critically — take the highest severity band across the resolved tokens so a composite string containing one critical defect is never routed as minor.

_DELIMS = re.compile(r"[_+\-;,/]+")
_BAND_RANK = {"minor": 0, "major": 1, "critical": 2}


def resolve_composite(raw: str, reg: DefectRegistry) -> list[Resolution]:
    tokens = [t for t in _DELIMS.split(raw) if t.strip()]
    # A single-token string is just the scalar path; keep the delimiters that
    # are legitimately part of a code (e.g. FRT-014) by trying the whole string first.
    whole = resolve(raw, reg)
    if whole.stage != "quarantine":
        return [whole]
    return [resolve(tok, reg) for tok in tokens] or [whole]


def worst_band(resolutions: list[Resolution]) -> str | None:
    bands = [r.defect.severity_band for r in resolutions if r.defect]
    return max(bands, key=_BAND_RANK.__getitem__) if bands else None

Every resolution is logged with the raw input, the matched canonical code, the resolution stage, and the confidence — this is the evidence a DOT auditor reads back. When the worst resolved band is critical, trigger the OOS evaluation; when any token quarantines, hold the whole record rather than acting on a partial result.

def decide(raw: str, reg: DefectRegistry) -> dict:
    resolutions = resolve_composite(raw, reg)
    quarantined = [r for r in resolutions if r.defect is None]
    band = worst_band(resolutions)

    if quarantined:
        action = "quarantine_hold"           # human review before any dispatch decision
    elif band == "critical":
        action = "trigger_oos_hold"          # 49 CFR 396.11(c)(2): unresolved critical defect
    elif band == "major":
        action = "open_repair_order"
    else:
        action = "schedule_maintenance"

    return {
        "raw_input": raw,
        "canonical": [r.defect.canonical for r in resolutions if r.defect],
        "severity_band": band,
        "stages": [r.stage for r in resolutions],
        "min_confidence": min(r.confidence for r in resolutions),
        "action": action,
        "registry_version": reg.version,     # pin the decision to the config that produced it
    }

Records that clear resolution are handed to Critical vs Non-Critical Routing Logic for dispatch action, while high-volume batches flow through Async Batching for High-Volume Ingestion. The upstream free-text cleanup that feeds raw_code is covered in Normalizing Inconsistent Driver Input Fields.

The resolver is a compliance component, so correctness is asserted, not assumed. Three test families matter.

Determinism and ladder ordering. Assert that a string matching both an alias and a pattern resolves via exact, proving the ladder order holds:

def test_exact_wins_over_fuzzy(reg):
    # "ooa" is an exact alias; "ooaa" (one insertion) must fall to fuzzy, not exact.
    assert resolve("ooa", reg).stage == "exact"
    assert resolve("ooaa", reg).stage == "fuzzy"


def test_quarantine_never_maps_a_compliance_action(reg):
    r = decide("plasma conduit misaligned", reg)   # nonsense: no confident match
    assert r["action"] == "quarantine_hold"
    assert r["canonical"] == []

Composite severity is a max, not a first-match. The single most important invariant — a composite that contains a critical defect must escalate:

def test_composite_takes_worst_band(reg):
    r = decide("headlight out+brakes out of adjustment", reg)
    assert r["severity_band"] == "critical"
    assert r["action"] == "trigger_oos_hold"

Confidence-floor contract. Any resolution below the fuzzy floor must not carry confidence == 1.0; assert the boundary so a future edit that widens FUZZY_MAX_EDITS fails a test rather than silently loosening a safety decision. Pair these with a golden-file corpus of real anonymized raw_code values from each OEM and legacy vendor, re-running the full corpus on every registry version bump so a mapping regression is caught before deploy.

• Delimiter collision with legitimate code characters. A canonical vendor code like FRT-014 contains a hyphen, which is also a composite delimiter. Splitting first would shred it into FRT and 014. The fix is in Step 4: attempt whole-string resolution before delimiter splitting, and only decompose when the whole string quarantines. Test both FRT-014 (single) and LIGHTS-BRAKES (composite) explicitly.

• Fuzzy matching bridging distinct defects. With a two-edit tolerance, short codes are dangerous: BRK and BLK are one edit apart but mean brakes versus a blackout lamp. Bounded edit distance alone is not enough for tokens under ~5 characters. Suppress fuzzy matching for short tokens and require them to hit an exact alias or quarantine, so proximity never crosses a system-domain boundary into an OOS-relevant code.

• Unicode homoglyphs and locale casing. Mobile keyboards inject non-breaking spaces, curly quotes, and Turkish dotless-i (ı), which break naive lower() comparisons — the Turkish-i problem means İ.lower() is not i under some locales. Always normalize with unicodedata.normalize("NFC", ...) before lowercasing (as in Step 2), pin normalization to NFC not NFKC to avoid collapsing meaningful characters, and treat casing as invariant rather than locale-dependent.

• Silent registry drift. When an OEM ships new firmware, unmapped codes climb. If the quarantine rate is unmonitored, coverage erodes invisibly and real defects pile up in a queue nobody watches. Emit the quarantine ratio per registry version as a metric and alert when it crosses a threshold (e.g. > 2% of daily volume), treating a spike as a mapping-coverage incident, not a data-quality footnote.

• Defect Taxonomy Mapping for Heavy Trucks — the canonical code hierarchy and severity bands this resolver maps into.
• Standardized DVIR JSON Schema Design — the validated contract that supplies the raw_code field.
• Normalizing Inconsistent Driver Input Fields — upstream free-text cleanup that precedes code resolution.
• Critical vs Non-Critical Routing Logic — what happens to a resolved critical defect downstream.
• Async Batching for High-Volume Ingestion — running the resolver over telematics bursts.

Back to Defect Taxonomy Mapping for Heavy Trucks, part of the Core DVIR Architecture & FMCSA Compliance Mapping reference architecture.