Quality Control Workflow
Overview
Automate construction quality control workflows from inspection planning through defect resolution. Track quality metrics, manage non-conformance reports (NCRs), and ensure specification compliance.
"Structured QC workflows reduce rework by 40% and improve first-time quality" — DDC Community
QC Workflow Stages
┌─────────────────────────────────────────────────────────────────┐
│ QC WORKFLOW PIPELINE │
├─────────────────────────────────────────────────────────────────┤
│ │
│ Plan → Inspect → Document → Resolve │
│ ──── ─────── ──────── ─────── │
│ 📋 ITP 🔍 Check 📝 NCR 🔧 Fix │
│ 📅 Schedule 📸 Photo 📊 Log ✅ Verify │
│ 👥 Assign 📏 Measure 📧 Notify 📈 Close │
│ │
└─────────────────────────────────────────────────────────────────┘Technical Implementation
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from enum import Enum
from datetime import datetime, timedelta
import json
class QCStatus(Enum):
PENDING = "pending"
IN_PROGRESS = "in_progress"
PASSED = "passed"
FAILED = "failed"
ON_HOLD = "on_hold"
class DefectSeverity(Enum):
CRITICAL = "critical" # Stop work, structural/safety issue
MAJOR = "major" # Must fix before cover-up
MINOR = "minor" # Fix before completion
COSMETIC = "cosmetic" # Punch list item
class NCRStatus(Enum):
DRAFT = "draft"
ISSUED = "issued"
ACCEPTED = "accepted"
DISPUTED = "disputed"
IN_REMEDIATION = "in_remediation"
VERIFIED = "verified"
CLOSED = "closed"
@dataclass
class InspectionPoint:
id: str
name: str
spec_reference: str
acceptance_criteria: str
inspection_method: str
hold_point: bool = False # Requires sign-off before proceeding
witness_point: bool = False # Client/engineer may witness
@dataclass
class QCInspection:
id: str
inspection_point: InspectionPoint
location: str
scheduled_date: datetime
inspector: str
status: QCStatus = QCStatus.PENDING
actual_date: Optional[datetime] = None
result: str = ""
measurements: Dict[str, float] = field(default_factory=dict)
photos: List[str] = field(default_factory=list)
notes: str = ""
sign_off_by: str = ""
@dataclass
class Defect:
id: str
inspection_id: str
description: str
location: str
severity: DefectSeverity
spec_reference: str
photos: List[str] = field(default_factory=list)
assigned_to: str = ""
due_date: Optional[datetime] = None
status: str = "open"
root_cause: str = ""
corrective_action: str = ""
verified_by: str = ""
closed_date: Optional[datetime] = None
@dataclass
class NonConformanceReport:
id: str
defect_ids: List[str]
title: str
description: str
contractor: str
spec_sections: List[str]
status: NCRStatus = NCRStatus.DRAFT
issued_date: Optional[datetime] = None
response_due: Optional[datetime] = None
contractor_response: str = ""
remediation_plan: str = ""
cost_impact: float = 0.0
schedule_impact_days: int = 0
class QualityControlManager:
"""Manage construction quality control workflow."""
def __init__(self, project_id: str, project_name: str):
self.project_id = project_id
self.project_name = project_name
self.inspection_points: Dict[str, InspectionPoint] = {}
self.inspections: Dict[str, QCInspection] = {}
self.defects: Dict[str, Defect] = {}
self.ncrs: Dict[str, NonConformanceReport] = {}
def create_itp(self, work_package: str, inspection_points: List[Dict]) -> List[InspectionPoint]:
"""Create Inspection and Test Plan (ITP)."""
created = []
for idx, point in enumerate(inspection_points):
ip = InspectionPoint(
id=f"{work_package}-{idx+1:03d}",
name=point['name'],
spec_reference=point.get('spec_ref', ''),
acceptance_criteria=point.get('criteria', ''),
inspection_method=point.get('method', 'Visual'),
hold_point=point.get('hold_point', False),
witness_point=point.get('witness_point', False)
)
self.inspection_points[ip.id] = ip
created.append(ip)
return created
def schedule_inspection(self, inspection_point_id: str, location: str,
scheduled_date: datetime, inspector: str) -> QCInspection:
"""Schedule a QC inspection."""
if inspection_point_id not in self.inspection_points:
raise ValueError(f"Inspection point {inspection_point_id} not found")
inspection_id = f"QCI-{datetime.now().strftime('%Y%m%d%H%M%S')}"
inspection = QCInspection(
id=inspection_id,
inspection_point=self.inspection_points[inspection_point_id],
location=location,
scheduled_date=scheduled_date,
inspector=inspector
)
self.inspections[inspection_id] = inspection
return inspection
def conduct_inspection(self, inspection_id: str, result: str,
measurements: Dict[str, float] = None,
photos: List[str] = None,
notes: str = "") -> QCInspection:
"""Record inspection results."""
if inspection_id not in self.inspections:
raise ValueError(f"Inspection {inspection_id} not found")
inspection = self.inspections[inspection_id]
inspection.actual_date = datetime.now()
inspection.result = result
inspection.measurements = measurements or {}
inspection.photos = photos or []
inspection.notes = notes
inspection.status = QCStatus.PASSED if result == "pass" else QCStatus.FAILED
return inspection
def record_defect(self, inspection_id: str, description: str,
location: str, severity: DefectSeverity,
spec_reference: str, photos: List[str] = None,
assigned_to: str = "") -> Defect:
"""Record a defect from inspection."""
defect_id = f"DEF-{datetime.now().strftime('%Y%m%d%H%M%S')}"
# Set due date based on severity
due_days = {
DefectSeverity.CRITICAL: 1,
DefectSeverity.MAJOR: 3,
DefectSeverity.MINOR: 7,
DefectSeverity.COSMETIC: 14
}
defect = Defect(
id=defect_id,
inspection_id=inspection_id,
description=description,
location=location,
severity=severity,
spec_reference=spec_reference,
photos=photos or [],
assigned_to=assigned_to,
due_date=datetime.now() + timedelta(days=due_days[severity])
)
self.defects[defect_id] = defect
return defect
def create_ncr(self, defect_ids: List[str], title: str,
contractor: str, description: str = "") -> NonConformanceReport:
"""Create Non-Conformance Report from defects."""
ncr_id = f"NCR-{datetime.now().strftime('%Y%m%d%H%M%S')}"
# Collect spec references from defects
spec_sections = []
for def_id in defect_ids:
if def_id in self.defects:
spec_sections.append(self.defects[def_id].spec_reference)
ncr = NonConformanceReport(
id=ncr_id,
defect_ids=defect_ids,
title=title,
description=description or self._generate_ncr_description(defect_ids),
contractor=contractor,
spec_sections=list(set(spec_sections)),
response_due=datetime.now() + timedelta(days=5)
)
self.ncrs[ncr_id] = ncr
return ncr
def _generate_ncr_description(self, defect_ids: List[str]) -> str:
"""Generate NCR description from defects."""
lines = ["The following non-conformances have been identified:", ""]
for def_id in defect_ids:
if def_id in self.defects:
d = self.defects[def_id]
lines.append(f"- {d.description} at {d.location}")
lines.append(f" Reference: {d.spec_reference}")
return "\n".join(lines)
def issue_ncr(self, ncr_id: str) -> NonConformanceReport:
"""Issue NCR to contractor."""
if ncr_id not in self.ncrs:
raise ValueError(f"NCR {ncr_id} not found")
ncr = self.ncrs[ncr_id]
ncr.status = NCRStatus.ISSUED
ncr.issued_date = datetime.now()
return ncr
def record_ncr_response(self, ncr_id: str, response: str,
remediation_plan: str, cost_impact: float = 0,
schedule_impact: int = 0) -> NonConformanceReport:
"""Record contractor response to NCR."""
if ncr_id not in self.ncrs:
raise ValueError(f"NCR {ncr_id} not found")
ncr = self.ncrs[ncr_id]
ncr.contractor_response = response
ncr.remediation_plan = remediation_plan
ncr.cost_impact = cost_impact
ncr.schedule_impact_days = schedule_impact
ncr.status = NCRStatus.ACCEPTED
return ncr
def verify_remediation(self, ncr_id: str, verified_by: str) -> NonConformanceReport:
"""Verify NCR remediation is complete."""
if ncr_id not in self.ncrs:
raise ValueError(f"NCR {ncr_id} not found")
ncr = self.ncrs[ncr_id]
ncr.status = NCRStatus.VERIFIED
# Close associated defects
for def_id in ncr.defect_ids:
if def_id in self.defects:
self.defects[def_id].status = "closed"
self.defects[def_id].verified_by = verified_by
self.defects[def_id].closed_date = datetime.now()
return ncr
def get_quality_metrics(self) -> Dict:
"""Calculate quality metrics."""
total_inspections = len(self.inspections)
passed = len([i for i in self.inspections.values() if i.status == QCStatus.PASSED])
failed = len([i for i in self.inspections.values() if i.status == QCStatus.FAILED])
open_defects = len([d for d in self.defects.values() if d.status == "open"])
overdue_defects = len([d for d in self.defects.values()
if d.status == "open" and d.due_date and d.due_date < datetime.now()])
open_ncrs = len([n for n in self.ncrs.values()
if n.status not in [NCRStatus.VERIFIED, NCRStatus.CLOSED]])
return {
"total_inspections": total_inspections,
"passed_inspections": passed,
"failed_inspections": failed,
"first_time_pass_rate": (passed / total_inspections * 100) if total_inspections else 0,
"open_defects": open_defects,
"overdue_defects": overdue_defects,
"open_ncrs": open_ncrs,
"total_cost_impact": sum(n.cost_impact for n in self.ncrs.values()),
"total_schedule_impact": sum(n.schedule_impact_days for n in self.ncrs.values())
}
def generate_qc_report(self) -> str:
"""Generate QC status report."""
metrics = self.get_quality_metrics()
lines = [
f"# Quality Control Report",
f"",
f"**Project:** {self.project_name}",
f"**Date:** {datetime.now().strftime('%Y-%m-%d')}",
f"",
f"## Summary Metrics",
f"",
f"| Metric | Value |",
f"|--------|-------|",
f"| First-Time Pass Rate | {metrics['first_time_pass_rate']:.1f}% |",
f"| Open Defects | {metrics['open_defects']} |",
f"| Overdue Defects | {metrics['overdue_defects']} |",
f"| Open NCRs | {metrics['open_ncrs']} |",
f"| Cost Impact | ${metrics['total_cost_impact']:,.0f} |",
f"| Schedule Impact | {metrics['total_schedule_impact']} days |",
f"",
]
# Critical defects
critical = [d for d in self.defects.values()
if d.severity == DefectSeverity.CRITICAL and d.status == "open"]
if critical:
lines.append("## Critical Open Defects")
for d in critical:
lines.append(f"- **{d.id}**: {d.description}")
lines.append(f" - Location: {d.location}")
lines.append(f" - Due: {d.due_date.strftime('%Y-%m-%d') if d.due_date else 'N/A'}")
return "\n".join(lines)Quick Start
# Initialize QC Manager
qc = QualityControlManager("PRJ-001", "Office Tower")
# Create Inspection Test Plan
concrete_itp = qc.create_itp("CONC", [
{"name": "Rebar placement", "spec_ref": "03200", "criteria": "Per drawings", "hold_point": True},
{"name": "Concrete placement", "spec_ref": "03300", "criteria": "Slump 4-6 inches"},
{"name": "Concrete finish", "spec_ref": "03350", "criteria": "Level within 1/8 inch"},
])
# Schedule inspection
inspection = qc.schedule_inspection(
"CONC-001",
location="Level 3, Grid A-C",
scheduled_date=datetime.now(),
inspector="QC Engineer"
)
# Conduct inspection (found defect)
qc.conduct_inspection(inspection.id, result="fail", notes="Rebar spacing incorrect")
# Record defect
defect = qc.record_defect(
inspection.id,
description="Rebar spacing 8 inches instead of 6 inches per spec",
location="Level 3, Grid B-2",
severity=DefectSeverity.MAJOR,
spec_reference="03200-3.2",
assigned_to="ABC Concrete"
)
# Create NCR
ncr = qc.create_ncr([defect.id], "Rebar Spacing Non-Conformance", "ABC Concrete")
qc.issue_ncr(ncr.id)
# Get metrics
print(qc.generate_qc_report())Requirements
pip install (no external dependencies)