10.2 Building a Basic Automated QA Pipeline

Introduction

10.1 built a batch runner that opens a folder of IFC files, extracts data from every wall it finds, and honestly records which files it could and couldn't process. What it doesn't do is look at the extracted values and say anything about whether they're any good. A wall with a net_volume of 0 gets treated identically to a wall with a perfectly sensible one, both are just numbers sitting in a CSV. This lesson adds that missing layer: a small set of explicit, hand-written rules that look at each extracted result and flag something worth a person's attention.

There are two different questions running through this pipeline, worth separating clearly before writing any check. Extraction asks: what does the IFC file actually contain. QA asks: does what was extracted satisfy a rule the project cares about. An extraction function reports what's genuinely in the file, nothing more. A QA check makes a judgment against a rule someone wrote. The same extracted value can be a completely accurate, valid piece of data and still fail a project-specific QA rule, and keeping those two questions distinct is what makes the rest of this lesson make sense.

What a Check Means at This Level

A "check" here is deliberately simple: one clear, explicit rule, applied to one extracted value, returning a result and a reason. Is a required value present at all. Does a numeric value satisfy a specific project rule. Is a value the type it's supposed to be. This is not the depth Module 6 covered in buildingSMART's own multi-stage validation, and it's not IDS from Module 7, a standardized, machine-interoperable specification format that produces the same result in any compliant tool. This is the plainest version possible: rules written directly in Python, specific to whatever a project's own pipeline needs to check right now, nothing more formal than that.

Worth stating plainly, since it's an easy thing to assume without noticing: a passing result here means this particular rule passed, nothing broader. It doesn't mean the IFC file is correct, or complete, or validated in any formal sense. The value of the conclusion is bounded entirely by the quality and coverage of the rules someone actually wrote.

Structuring a Check as a Function

Following the same discipline Module 9 built extraction around, one check gets its own small function, taking a single extracted result and returning a structured verdict rather than printing anything directly. A check genuinely has three possible outcomes, not two, it can pass, fail, or have nothing to check at all, and the result structure should say which:

def check_volume_present(row):
    if row["net_volume"] is None:
        return {"check": "volume_present", "status": "failed", "reason": "NetVolume is missing or unset"}
    return {"check": "volume_present", "status": "passed", "reason": None}

def check_volume_positive(row):
    volume = row["net_volume"]
    if volume is None:
        return {"check": "volume_positive", "status": "skipped", "reason": "NetVolume is unavailable"}
    if volume <= 0:
        return {"check": "volume_positive", "status": "failed", "reason": f"NetVolume is {volume}, not a positive number"}
    return {"check": "volume_positive", "status": "passed", "reason": None}

Two small, independent functions, each checking exactly one thing. check_volume_present only cares whether a value exists at all. check_volume_positive only cares whether an existing volume is greater than zero, a deliberately simple project rule chosen for this example, not a universal claim about what's valid for every IFC quantity or element type. When there's nothing to check, it reports skipped, not passed, since those genuinely mean different things and a QA report that collapses them together loses real information. A missing value is already check_volume_present's job to flag; check_volume_positive skipping rather than failing on the same missing value avoids reporting the same underlying problem twice.

Running Multiple Checks and Collecting Results

A list of check functions applied to every row of extracted data turns into a structured report rather than scattered print statements:

checks = [check_volume_present, check_volume_positive]

def run_checks(row):
    return [check(row) for check in checks]

def flatten_check_results(rows):
    flagged = []
    for row in rows:
        for result in run_checks(row):
            if result["status"] == "failed":
                flagged.append({
                    "wall_name": row["wall_name"],
                    "global_id": row["global_id"],
                    "check": result["check"],
                    "reason": result["reason"],
                })
    return flagged

Adding a new check later means writing one more small function and adding it to the checks list, nothing else in this structure needs to change. flatten_check_results keeps only genuine failures, filtering explicitly on status == "failed" rather than just not passed, since a skipped check shouldn't show up in a report meant to surface actual problems.

A Complete Pipeline Script

Combining this with 10.1's batch runner: extract, check, and write out both the raw data and a focused list of flagged issues. The following assumes run_batch() and the extraction functions from 10.1 are already available in the same script.

from pathlib import Path
import csv

def check_volume_present(row):
    if row["net_volume"] is None:
        return {"check": "volume_present", "status": "failed", "reason": "NetVolume is missing or unset"}
    return {"check": "volume_present", "status": "passed", "reason": None}

def check_volume_positive(row):
    volume = row["net_volume"]
    if volume is None:
        return {"check": "volume_positive", "status": "skipped", "reason": "NetVolume is unavailable"}
    if volume <= 0:
        return {"check": "volume_positive", "status": "failed", "reason": f"NetVolume is {volume}, not a positive number"}
    return {"check": "volume_positive", "status": "passed", "reason": None}

checks = [check_volume_present, check_volume_positive]

def run_checks(row):
    return [check(row) for check in checks]

def flatten_check_results(rows):
    flagged = []
    for row in rows:
        for result in run_checks(row):
            if result["status"] == "failed":
                flagged.append({
                    "wall_name": row["wall_name"],
                    "global_id": row["global_id"],
                    "check": result["check"],
                    "reason": result["reason"],
                })
    return flagged

def write_csv(rows, output_path, fieldnames):
    with open(output_path, "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(rows)

# run_batch() comes unchanged from Lesson 10.1
run_log, all_results = run_batch("/path/to/ifc_folder")
flagged_issues = flatten_check_results(all_results)

write_csv(run_log, "run_log.csv", ["file", "status", "error"])
write_csv(all_results, "wall_quantities.csv", ["file", "wall_name", "global_id", "net_volume"])
write_csv(flagged_issues, "flagged_issues.csv", ["wall_name", "global_id", "check", "reason"])

print(f"{len(flagged_issues)} issues flagged across {len(all_results)} walls")

Three files come out of this run: which files processed successfully, every extracted value, and a short, focused list of exactly what needs a second look. write_csv() now takes explicit fieldnames rather than reading them off the first row, so flagged_issues.csv still gets created with proper headers even on a clean run with zero issues, a report that says "nothing was flagged" rather than a file that simply doesn't exist. run_batch() here is the same function built in 10.1, unchanged, since the checking layer sits entirely on top of extraction rather than needing to touch it.

The full shape this pipeline follows is worth holding onto as one picture: can the file be opened and processed at all, tracked in the run log; then, for whatever was successfully extracted, does it satisfy the rules that were written, tracked in the flagged issues. Two different questions, two different failure surfaces, kept clearly apart.

What This Is, and Isn't

Worth being direct about this rather than letting it go unsaid. What's built here is genuinely useful for a project's own internal, ad hoc checks, quick, specific, and easy to extend. It is not IDS, and it isn't trying to be. IDS addresses a real limitation of hand-written, project-specific checks like these: the requirement and the logic that checks it are tied to one particular codebase, understood only by whoever can read that specific script, rather than expressed as a standardized, machine-readable specification any compliant tool can interpret the same way. IDS solves a broader interoperability problem than "custom scripts are bad", it lets a requirement be written once and checked identically across different tools and people, not just the one script that happens to implement it today.

Neither approach is simply better. Hand-rolled checks like the ones in this lesson are fast to write and perfectly suited to a project's own internal, informal use. The moment a check needs to be shared across tools, understood by someone who doesn't read Python, or trusted to produce the same result regardless of which software runs it, that's the moment IDS, covered in Module 7, is the right tool instead.

What's Next

Everything in this module so far has been code, run manually, by a person choosing to execute it. The next lesson steps back from code entirely and looks at where automation like this actually fits inside a real project, specifically within a BIM Execution Plan, the document Module 8 already introduced as the delivery team's response to an EIR.