How to Map CSI MasterFormat to Custom WBS Codes in Python

Construction pipelines ingest cost codes, RFIs, and submittals tagged with CSI MasterFormat divisions, but no two source systems format those tags the same way. This page solves one precise problem: how to deterministically resolve a raw MasterFormat string to an enterprise-specific Work Breakdown Structure code without silent data loss, so that a misformatted division number never posts cost against the wrong account. The resolver normalizes inconsistent inputs to a canonical six-digit number, looks it up against an enterprise mapping table, and emits a routing decision — auto-commit, human review, or quarantine — backed by a confidence score and an audit record. It sits inside a deterministic construction data architecture and taxonomy and is the operational heart of any practical WBS mapping strategy: the WBS node names where work sits in scope, while the MasterFormat number names what the work is, and this layer is the bridge between the two. It targets Python automation builders who need predictable cost routing under real-world input variance from ERP exports, Procore, and CSV cost reports.

Key Rules and Specification

A mapping engine is only as reliable as the constraints it enforces at the boundary. The following rules govern every resolution:

Canonical MasterFormat shape. The 2018/2020 editions use a six-digit hierarchy rendered as DD SS NN — Division, BroadScope, NarrowScope (for example 03 30 00). Legacy 2004 exports carry five digits (DD SSS) and must be padded, not rejected.
Normalize before lookup. Strip vendor suffixes (-ALT1, _REV, -01), collapse every delimiter style (spaces, hyphens, or none) to the canonical form, and only then perform a dictionary lookup. Direct string matching on raw input fails across systems.
WBS element pattern. Enterprise WBS codes follow a fixed PROJ-NNN-DIV-NN contract and are validated by regex before they leave the resolver, so a malformed target can never enter the cost ledger.
Discipline is a closed set. Every resolved node carries a discipline drawn from a Literal of ARCH, STR, MEP, CIV, ELEC, PLMB — never a free string — so downstream rollups cannot split on a typo.
Confidence bands drive routing. Scores are site-canonical: 0.92 and above auto-routes, 0.75 to 0.92 flags for human review, and below 0.75 quarantines the record. Division 00 (procurement/contracting) and 01 (general requirements) are cross-disciplinary and route to a project-overhead bucket rather than a discipline node.
Idempotent and traceable. Normalization is a pure function: re-running it on its own output is a no-op, so a broker retry produces the identical code. Every resolution preserves the original input for the audit trail.

Match tier	Trigger	Confidence	Route state
`exact`	Six-digit code in the enterprise map	0.99	auto_route
`cross_division`	Division `00`/`01` → overhead bucket	0.93	auto_route
`division`	Only the parent division is mapped	0.83	human_review
`quarantine`	No mapping, or normalization failed	0.00	quarantine

Production Code Example

The resolver below uses Pydantic v2 to make the output of a mapping a validated, frozen contract. Construction-domain constants — the MasterFormat number, the WBS pattern, the discipline set — are regex- or Literal-constrained fields, so an invalid resolution raises at construction time rather than corrupting a rollup downstream. Pattern-matching details follow the standard Python re module documentation.

from __future__ import annotations

import logging
import re
from decimal import Decimal
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field, field_validator

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)-8s | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger("csi_wbs")

# Canonical six-digit MasterFormat number, e.g. "03 30 00" (2018/2020 editions).
MASTERFORMAT_RE = re.compile(r"^\d{2} \d{2} \d{2}$")
# Enterprise WBS element contract: PROJ-NNN-DIV-NN, e.g. "TWR-014-STR-03".
WBS_RE = re.compile(r"^[A-Z]{2,5}-\d{3}-[A-Z]{2,4}-\d{2}$")
# Vendor suffix tail: -ALT1, _REV, -PHASE2 (alphabetic-led only — never numeric).
SUFFIX_RE = re.compile(r"[-_\s][A-Za-z]\w*$")

Discipline = Literal["ARCH", "STR", "MEP", "CIV", "ELEC", "PLMB"]
MatchType = Literal["exact", "cross_division", "division", "quarantine"]
RouteState = Literal["auto_route", "human_review", "quarantine"]

AUTO_ROUTE = Decimal("0.92")   # site-canonical: >= 0.92 commits automatically
HUMAN_REVIEW = Decimal("0.75")  # 0.75–0.92 holds for an estimator to confirm


class WBSMapping(BaseModel):
    """A validated CSI->WBS resolution that carries its own routing decision."""

    model_config = ConfigDict(frozen=True)

    source_code: str = Field(..., description="Raw inbound code, preserved for audit")
    masterformat: str = Field(..., description="Canonical six-digit MasterFormat number")
    wbs_code: str
    discipline: Discipline
    match_type: MatchType
    confidence: Decimal = Field(..., ge=0, le=1)

    @field_validator("masterformat")
    @classmethod
    def _canonical_masterformat(cls, v: str) -> str:
        if not MASTERFORMAT_RE.match(v):
            raise ValueError(f"Non-canonical MasterFormat number: {v!r}")
        return v

    @field_validator("wbs_code")
    @classmethod
    def _valid_wbs(cls, v: str) -> str:
        if not WBS_RE.match(v):
            raise ValueError(f"WBS code violates PROJ-NNN-DIV-NN: {v!r}")
        return v

    @property
    def route_state(self) -> RouteState:
        if self.confidence >= AUTO_ROUTE:
            return "auto_route"
        if self.confidence >= HUMAN_REVIEW:
            return "human_review"
        return "quarantine"


def normalize_masterformat(raw: str) -> str:
    """Collapse any delimiter/suffix style to canonical 'DD SS NN'. Idempotent."""
    stripped = SUFFIX_RE.sub("", raw.strip())          # drop -ALT1, _REV, ...
    digits = re.sub(r"\D", "", stripped)               # keep digits only
    if len(digits) == 5:                               # 2004 five-digit form
        digits += "0"                                  # pad NarrowScope -> six digits
    if len(digits) != 6:
        raise ValueError(f"Expected 5 or 6 MasterFormat digits in {raw!r}")
    return f"{digits[0:2]} {digits[2:4]} {digits[4:6]}"


class CSIToWBSMapper:
    """Deterministic, audit-friendly CSI MasterFormat -> enterprise WBS resolver."""

    def __init__(
        self,
        exact_map: dict[str, str],
        division_map: dict[str, str],
        division_discipline: dict[str, Discipline],
        overhead_wbs: str,
        quarantine_wbs: str,
    ) -> None:
        self.exact_map = exact_map
        self.division_map = division_map
        self.division_discipline = division_discipline
        self.overhead_wbs = overhead_wbs
        self.quarantine_wbs = quarantine_wbs

    def _discipline_for(self, division: str) -> Discipline:
        # Default unknown divisions to MEP-adjacent review rather than guessing.
        return self.division_discipline.get(division, "MEP")

    def resolve(self, raw_code: str) -> WBSMapping:
        try:
            canonical = normalize_masterformat(raw_code)
        except ValueError as exc:
            logger.warning("Normalization failed, quarantining: %s", exc)
            return WBSMapping(
                source_code=raw_code,
                masterformat="00 00 00",
                wbs_code=self.quarantine_wbs,
                discipline="MEP",
                match_type="quarantine",
                confidence=Decimal("0.0"),
            )

        division = canonical[:2]

        # 1. Exact six-digit match — highest confidence, auto-routes.
        if canonical in self.exact_map:
            return WBSMapping(
                source_code=raw_code, masterformat=canonical,
                wbs_code=self.exact_map[canonical],
                discipline=self._discipline_for(division),
                match_type="exact", confidence=Decimal("0.99"),
            )

        # 2. Division 00/01 are cross-disciplinary -> deterministic overhead bucket.
        if division in ("00", "01"):
            return WBSMapping(
                source_code=raw_code, masterformat=canonical,
                wbs_code=self.overhead_wbs, discipline="MEP",
                match_type="cross_division", confidence=Decimal("0.93"),
            )

        # 3. Only the parent division is mapped -> roll up, but ask a human.
        if division in self.division_map:
            return WBSMapping(
                source_code=raw_code, masterformat=canonical,
                wbs_code=self.division_map[division],
                discipline=self._discipline_for(division),
                match_type="division", confidence=Decimal("0.83"),
            )

        # 4. Nothing matched -> quarantine for manual triage, never silent-drop.
        logger.warning("No mapping for %s; quarantining", canonical)
        return WBSMapping(
            source_code=raw_code, masterformat=canonical,
            wbs_code=self.quarantine_wbs,
            discipline=self._discipline_for(division),
            match_type="quarantine", confidence=Decimal("0.10"),
        )

In a live batch — feeding the resolver from an ERP or Procore export — isolate each row so one bad code never aborts the run, and emit a structured audit line for everything that does not auto-route. The logging configuration above writes those lines per the Python Logging HOWTO; model_dump_json() serializes each resolution straight into the audit store.

def map_cost_codes(mapper: CSIToWBSMapper, raw_codes: list[str]) -> list[WBSMapping]:
    """Batch-resolve raw CSI strings with per-row error isolation and audit logging."""
    results: list[WBSMapping] = []
    for code in raw_codes:
        mapping = mapper.resolve(code.strip())
        results.append(mapping)
        if mapping.route_state != "auto_route":
            logger.info("AUDIT | %s", mapping.model_dump_json())
    return results


if __name__ == "__main__":
    mapper = CSIToWBSMapper(
        exact_map={"03 30 00": "TWR-014-STR-03", "26 05 19": "TWR-014-ELEC-07"},
        division_map={"03": "TWR-014-STR-00", "09": "TWR-014-ARCH-00"},
        division_discipline={"03": "STR", "09": "ARCH", "26": "ELEC",
                             "22": "PLMB", "23": "MEP", "31": "CIV"},
        overhead_wbs="TWR-014-MEP-99",
        quarantine_wbs="TWR-014-MEP-00",
    )
    samples = ["03 30 00", "033000-ALT1", "03 30 12", "01 00 00", "26-05-19", "ZZ"]
    for m in map_cost_codes(mapper, samples):
        print(f"{m.source_code:<14} -> {m.wbs_code:<16} {m.match_type:<14} {m.route_state}")

Common Mistakes and Gotchas

Suffix regex that eats trailing digits. A naive pattern like [-_]\w+$ will strip the -01 from a numeric revision identifier and from a legitimate 03 30 00-01 line item, silently changing which code you resolve. The pattern above ([-_\s][A-Za-z]\w*$) only removes suffixes that begin with a letter, so numeric tails survive normalization. If your organization appends numeric revisions, parse them into a separate field before normalizing rather than discarding them.

Routing unmapped codes to a real WBS bucket. The tempting shortcut is to send anything unrecognized to a catch-all cost account. That hides the failure: misrouted spend looks reconciled until audit. Quarantine instead — a low confidence score and a held record force triage, which is exactly how fallback alert routing is designed to surface the gap. Quarantine is the construction-automation cousin of a dead-letter queue.

Treating division-level rollups as exact. When only the parent division (03) is mapped and the granular code (03 30 12) is not, posting straight to the parent bucket at full confidence buries detail that estimating may need. Scoring that tier at 0.83 lands it in the human-review band, where an estimator either confirms the rollup or adds the missing six-digit mapping — the score is the signal, not just a number.

Where This Fits in the Pipeline

This resolver is one stage of the broader WBS mapping strategy under the construction data architecture and taxonomy. Upstream, a code only reaches it after it clears schema validation rules at the ingestion boundary; downstream, the resolved WBS code becomes the join key for budget code standardization, which aggregates committed cost against the matched node. Quarantined and human-review records hand off to error handling protocols so nothing exits the pipeline unresolved. The confidence bands used here are the same 0.92 / 0.75 thresholds applied across every routing decision on the site, which keeps audit behavior consistent from document intake to ledger.

Frequently Asked Questions

Why normalize to a space-delimited six-digit number instead of keeping hyphens?

Source systems use spaces, hyphens, or no separator at all, and they pad legacy codes differently, so raw string equality fails across systems. Collapsing every variant to one canonical DD SS NN form makes the lookup a clean dictionary hit and lets a single regex contract reject anything malformed at the boundary.

How are five-digit 2004 codes handled?

normalize_masterformat detects a five-digit DD SSS legacy code and pads the NarrowScope with a trailing zero to produce the canonical six digits (03300 becomes 03 30 00). Because the suffix stripper only removes alphabetic-led tails, it never consumes those trailing digits during the pad.

What happens to a code with no mapping?

It is quarantined, not routed to a real cost account. The resolver returns a low confidence score and a quarantine WBS bucket, which holds the record for manual triage and emits an audit line. Routing unmapped spend to a catch-all account is what makes misrouted cost look reconciled until audit.

How do the confidence bands map to action here?

They are site-canonical. An exact six-digit match (0.99) and a deterministic cross-division route (0.93) auto-route. A division-only rollup (0.83) falls in the 0.75–0.92 band and waits for an estimator to confirm. Anything below 0.75 quarantines.

Why is the discipline a Literal rather than a free string?

Cost rollups aggregate by discipline, and an exact set (ARCH, STR, MEP, CIV, ELEC, PLMB) means a typo can never silently create a phantom discipline bucket that splits a total. Pydantic rejects an unknown value at construction time, so the bad data never reaches the ledger.

← Back to WBS Mapping Strategies