关于 defect-detection-ai
用于施工质量控制的深度学习技能,可分析图像和视频以检测并分类结构及表面缺陷。它还可评估缺陷严重程度并生成检查报告。
主要功能
- 01检测图像和视频中的缺陷
- 02分类结构及表面缺陷
- 03评估缺陷严重程度
- 04生成检查报告
- 05支持模型训练
适用场景
- 01检查混凝土和钢结构
- 02从影像中查找裂缝及表面缺陷
- 03编制缺陷检查报告
- 04训练施工缺陷检测模型
安装
复制下面的命令,在终端中运行。
$
npx skills add https:/ / github.com/ datadrivenconstruction/ ddc_skills_for_ai_agents_in_construction --skill defect-detection-ai安装后在 Claude Code 中输入/defect-detection-ai确认已加载。
附带文件claw.jsoninstructions.md
详细介绍
AI 缺陷检测
概述
此技能通过深度学习实现施工质量控制中的缺陷检测。分析图像和视频,自动识别结构和表面缺陷、判定严重程度并生成检查报告。
可检测的缺陷:
- 混凝土:裂缝、剥落、蜂窝、泛碱
- 钢材:腐蚀、焊接缺陷、变形
- 砌体:砂浆劣化、位移
- 饰面:表面缺陷、涂层失效
- 机电:保温损坏、管道腐蚀
快速入门
import torch
import torch.nn as nn
from torchvision import transforms, models
from PIL import Image
from dataclasses import dataclass
from typing import List, Dict, Tuple
from enum import Enum
class DefectType(Enum):
CRACK = "crack"
SPALLING = "spalling"
CORROSION = "corrosion"
HONEYCOMBING = "honeycombing"
EFFLORESCENCE = "efflorescence"
DEFORMATION = "deformation"
SURFACE_DAMAGE = "surface_damage"
NO_DEFECT = "no_defect"
class SeverityLevel(Enum):
MINOR = "minor"
MODERATE = "moderate"
SEVERE = "severe"
CRITICAL = "critical"
@dataclass
class DefectDetection:
defect_type: DefectType
confidence: float
severity: SeverityLevel
bounding_box: Tuple[int, int, int, int] # x1, y1, x2, y2
area_ratio: float # Defect area as ratio of image
# Simple classifier using pretrained model
class SimpleDefectClassifier:
def __init__(self, num_classes: int = 8):
self.model = models.resnet18(pretrained=True)
self.model.fc = nn.Linear(self.model.fc.in_features, num_classes)
self.model.eval()
self.transform = transforms.Compose([
transforms.Resize((224, 224)),
transforms.ToTensor(),
transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])
])
self.classes = list(DefectType)
def predict(self, image_path: str) -> DefectDetection:
"""Classify defect in image"""
image = Image.open(image_path).convert('RGB')
input_tensor = self.transform(image).unsqueeze(0)
with torch.no_grad():
outputs = self.model(input_tensor)
probs = torch.softmax(outputs, dim=1)
confidence, predicted = torch.max(probs, 1)
defect_type = self.classes[predicted.item()]
return DefectDetection(
defect_type=defect_type,
confidence=confidence.item(),
severity=self._estimate_severity(confidence.item()),
bounding_box=(0, 0, image.width, image.height),
area_ratio=1.0
)
def _estimate_severity(self, confidence: float) -> SeverityLevel:
if confidence > 0.9:
return SeverityLevel.CRITICAL
elif confidence > 0.7:
return SeverityLevel.SEVERE
elif confidence > 0.5:
return SeverityLevel.MODERATE
else:
return SeverityLevel.MINOR
# Usage
classifier = SimpleDefectClassifier()
# result = classifier.predict("concrete_image.jpg")
# print(f"Defect: {result.defect_type.value}, Confidence: {result.confidence:.2%}")
综合缺陷检测系统
目标检测模型
import torch
import torch.nn as nn
from torchvision import transforms
from torchvision.models.detection import fasterrcnn_resnet50_fpn
from PIL import Image
import numpy as np
from dataclasses import dataclass, field
from typing import List, Dict, Tuple, Optional
from datetime import datetime
import json
@dataclass
class BoundingBox:
x1: int
y1: int
x2: int
y2: int
@property
def width(self) -> int:
return self.x2 - self.x1
@property
def height(self) -> int:
return self.y2 - self.y1
@property
def area(self) -> int:
return self.width * self.height
@property
def center(self) -> Tuple[int, int]:
return ((self.x1 + self.x2) // 2, (self.y1 + self.y2) // 2)
@dataclass
class DetectedDefect:
defect_id: str
defect_type: DefectType
confidence: float
severity: SeverityLevel
bounding_box: BoundingBox
area_sqm: Optional[float] = None
dimensions_mm: Optional[Tuple[float, float]] = None
metadata: Dict = field(default_factory=dict)
@dataclass
class InspectionResult:
inspection_id: str
image_path: str
timestamp: datetime
location: str
element_type: str
defects: List[DetectedDefect]
overall_condition: str
recommended_actions: List[str]
class DefectDetectionModel:
"""Deep learning defect detection with object detection"""
DEFECT_CLASSES = {
1: DefectType.CRACK,
2: DefectType.SPALLING,
3: DefectType.CORROSION,
4: DefectType.HONEYCOMBING,
5: DefectType.EFFLORESCENCE,
6: DefectType.DEFORMATION,
7: DefectType.SURFACE_DAMAGE
}
def __init__(self, model_path: str = None, device: str = 'cpu'):
self.device = torch.device(device)
# Initialize Faster R-CNN
self.model = fasterrcnn_resnet50_fpn(pretrained=True)
# Modify for our classes
num_classes = len(self.DEFECT_CLASSES) + 1 # +1 for background
in_features = self.model.roi_heads.box_predictor.cls_score.in_features
self.model.roi_heads.box_predictor = FastRCNNPredictor(in_features, num_classes)
if model_path:
self.model.load_state_dict(torch.load(model_path, map_location=self.device))
self.model.to(self.device)
self.model.eval()
self.transform = transforms.Compose([
transforms.ToTensor()
])
def detect(self, image_path: str, confidence_threshold: float = 0.5,
pixels_per_mm: float = None) -> List[DetectedDefect]:
"""Detect defects in image"""
image = Image.open(image_path).convert('RGB')
image_tensor = self.transform(image).to(self.device)
with torch.no_grad():
predictions = self.model([image_tensor])
pred = predictions[0]
defects = []
for i in range(len(pred['boxes'])):
score = pred['scores'][i].item()
if score < confidence_threshold:
continue
label = pred['labels'][i].item()
box = pred['boxes'][i].cpu().numpy()
defect_type = self.DEFECT_CLASSES.get(label, DefectType.SURFACE_DAMAGE)
bbox = BoundingBox(
x1=int(box[0]),
y1=int(box[1]),
x2=int(box[2]),
y2=int(box[3])
)
# Calculate dimensions if scale provided
dimensions_mm = None
if pixels_per_mm:
width_mm = bbox.width / pixels_per_mm
height_mm = bbox.height / pixels_per_mm
dimensions_mm = (width_mm, height_mm)
severity = self._classify_severity(defect_type, bbox, image.size)
defects.append(DetectedDefect(
defect_id=f"DEF-{i:04d}",
defect_type=defect_type,
confidence=score,
severity=severity,
bounding_box=bbox,
dimensions_mm=dimensions_mm
))
return defects
def _classify_severity(self, defect_type: DefectType,
bbox: BoundingBox,
image_size: Tuple[int, int]) -> SeverityLevel:
"""Classify defect severity based on type and size"""
image_area = image_size[0] * image_size[1]
defect_ratio = bbox.area / image_area
# Severity thresholds by defect type
thresholds = {
DefectType.CRACK: {'critical': 0.1, 'severe': 0.05, 'moderate': 0.02},
DefectType.SPALLING: {'critical': 0.15, 'severe': 0.08, 'moderate': 0.03},
DefectType.CORROSION: {'critical': 0.2, 'severe': 0.1, 'moderate': 0.05},
DefectType.HONEYCOMBING: {'critical': 0.1, 'severe': 0.05, 'moderate': 0.02},
DefectType.DEFORMATION: {'critical': 0.05, 'severe': 0.02, 'moderate': 0.01}
}
t = thresholds.get(defect_type, {'critical': 0.15, 'severe': 0.08, 'moderate': 0.03})
if defect_ratio >= t['critical']:
return SeverityLevel.CRITICAL
elif defect_ratio >= t['severe']:
return SeverityLevel.SEVERE
elif defect_ratio >= t['moderate']:
return SeverityLevel.MODERATE
else:
return SeverityLevel.MINOR
class FastRCNNPredictor(nn.Module):
"""Custom predictor for Faster R-CNN"""
def __init__(self, in_channels, num_classes):
super().__init__()
self.cls_score = nn.Linear(in_channels, num_classes)
self.bbox_pred = nn.Linear(in_channels, num_classes * 4)
def forward(self, x):
scores = self.cls_score(x)
bbox_deltas = self.bbox_pred(x)
return scores, bbox_deltas
裂缝分析系统
import cv2
import numpy as np
from typing import List, Tuple, Dict
class CrackAnalyzer:
"""Specialized crack detection and measurement"""
def __init__(self):
self.min_crack_length = 10 # pixels
self.min_crack_width = 2 # pixels
def detect_cracks(self, image_path: str,
pixels_per_mm: float = 1.0) -> List[Dict]:
"""Detect and measure cracks in image"""
# Load image
image = cv2.imread(image_path)
gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
# Enhance contrast
clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8))
enhanced = clahe.apply(gray)
# Edge detection
edges = cv2.Canny(enhanced, 50, 150)
# Morphological operations to connect crack segments
kernel = np.ones((3, 3), np.uint8)
dilated = cv2.dilate(edges, kernel, iterations=1)
closed = cv2.morphologyEx(dilated, cv2.MORPH_CLOSE, kernel)
# Find contours
contours, _ = cv2.findContours(closed, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
cracks = []
for i, contour in enumerate(contours):
# Filter by length
arc_length = cv2.arcLength(contour, False)
if arc_length < self.min_crack_length:
continue
# Get bounding box
x, y, w, h = cv2.boundingRect(contour)
# Calculate crack properties
length_px = arc_length
width_px = self._estimate_crack_width(gray, contour)
# Convert to mm
length_mm = length_px / pixels_per_mm
width_mm = width_px / pixels_per_mm
# Classify crack
crack_type = self._classify_crack(length_mm, width_mm, contour)
cracks.append({
'crack_id': f"CRACK-{i:04d}",
'type': crack_type,
'length_mm': length_mm,
'width_mm': width_mm,
'bounding_box': (x, y, x + w, y + h),
'contour': contour.tolist(),
'severity': self._get_crack_severity(width_mm, length_mm),
'orientation': self._get_crack_orientation(contour)
})
return cracks
def _estimate_crack_width(self, gray_image: np.ndarray,
contour: np.ndarray) -> float:
"""Estimate average crack width"""
# Create mask for contour
mask = np.zeros(gray_image.shape, dtype=np.uint8)
cv2.drawContours(mask, [contour], -1, 255, 1)
# Distance transform
dist = cv2.distanceTransform(mask, cv2.DIST_L2, 5)
# Get average distance (half-width)
nonzero = dist[dist > 0]
if len(nonzero) > 0:
return np.mean(nonzero) * 2
return 0
def _classify_crack(self, length_mm: float, width_mm: float,
contour: np.ndarray) -> str:
"""Classify crack type"""
# Fit line to get orientation
[vx, vy, x, y] = cv2.fitLine(contour, cv2.DIST_L2, 0, 0.01, 0.01)
angle = np.arctan2(vy, vx) * 180 / np.pi
if abs(angle) < 20 or abs(angle) > 160:
orientation = "horizontal"
elif 70 < abs(angle) < 110:
orientation = "vertical"
else:
orientation = "diagonal"
# Check for pattern (simplified)
if width_mm > 3:
return "structural_crack"
elif orientation == "horizontal" and length_mm > 100:
return "settlement_crack"
elif orientation == "diagonal":
return "shear_crack"
else:
return "shrinkage_crack"
def _get_crack_severity(self, width_mm: float, length_mm: float) -> str:
"""Determine crack severity based on dimensions"""
# Based on ACI 224R guidelines
if width_mm > 1.0:
return "critical"
elif width_mm > 0.4:
return "severe"
elif width_mm > 0.2:
return "moderate"
else:
return "minor"
def _get_crack_orientation(self, contour: np.ndarray) -> float:
"""Get crack orientation angle"""
[vx, vy, x, y] = cv2.fitLine(contour, cv2.DIST_L2, 0, 0.01, 0.01)
return float(np.arctan2(vy, vx) * 180 / np.pi)
def generate_crack_report(self, cracks: List[Dict]) -> Dict:
"""Generate summary report of detected cracks"""
if not cracks:
return {'message': 'No cracks detected'}
total_length = sum(c['length_mm'] for c in cracks)
max_width = max(c['width_mm'] for c in cracks)
severity_counts = {}
for c in cracks:
sev = c['severity']
severity_counts[sev] = severity_counts.get(sev, 0) + 1
return {
'total_cracks': len(cracks),
'total_length_mm': total_length,
'max_width_mm': max_width,
'avg_width_mm': sum(c['width_mm'] for c in cracks) / len(cracks),
'by_severity': severity_counts,
'by_type': self._group_by_type(cracks),
'most_severe': max(cracks, key=lambda c: c['width_mm'])
}
def _group_by_type(self, cracks: List[Dict]) -> Dict:
"""Group cracks by type"""
grouped = {}
for c in cracks:
t = c['type']
if t not in grouped:
grouped[t] = []
grouped[t].append(c['crack_id'])
return grouped
检查报告生成器
from datetime import datetime
import pandas as pd
class DefectInspectionSystem:
"""Complete defect inspection and reporting system"""
def __init__(self, detection_model: DefectDetectionModel):
self.model = detection_model
self.crack_analyzer = CrackAnalyzer()
self.inspections: List[InspectionResult] = []
def perform_inspection(self, image_path: str,
location: str,
element_type: str,
pixels_per_mm: float = None) -> InspectionResult:
"""Perform complete inspection on image"""
# Detect defects
defects = self.model.detect(image_path, pixels_per_mm=pixels_per_mm)
# Additional crack analysis for concrete
if element_type.lower() in ['concrete', 'slab', 'wall', 'column', 'beam']:
cracks = self.crack_analyzer.detect_cracks(image_path, pixels_per_mm or 1.0)
# Add detailed crack info to relevant defects
for defect in defects:
if defect.defect_type == DefectType.CRACK:
for crack in cracks:
# Check if crack overlaps with defect bbox
if self._boxes_overlap(defect.bounding_box, crack['bounding_box']):
defect.metadata['crack_details'] = crack
break
# Determine overall condition
overall_condition = self._assess_overall_condition(defects)
# Generate recommendations
recommendations = self._generate_recommendations(defects, element_type)
result = InspectionResult(
inspection_id=f"INS-{datetime.now().strftime('%Y%m%d%H%M%S')}",
image_path=image_path,
timestamp=datetime.now(),
location=location,
element_type=element_type,
defects=defects,
overall_condition=overall_condition,
recommended_actions=recommendations
)
self.inspections.append(result)
return result
def _boxes_overlap(self, box1: BoundingBox, box2: Tuple) -> bool:
"""Check if two bounding boxes overlap"""
x1_1, y1_1, x2_1, y2_1 = box1.x1, box1.y1, box1.x2, box1.y2
x1_2, y1_2, x2_2, y2_2 = box2
return not (x2_1 < x1_2 or x2_2 < x1_1 or y2_1 < y1_2 or y2_2 < y1_1)
def _assess_overall_condition(self, defects: List[DetectedDefect]) -> str:
"""Assess overall structural condition"""
if not defects:
return "Good"
severity_scores = {
SeverityLevel.MINOR: 1,
SeverityLevel.MODERATE: 2,
SeverityLevel.SEVERE: 3,
SeverityLevel.CRITICAL: 4
}
max_severity = max(severity_scores[d.severity] for d in defects)
total_defects = len(defects)
if max_severity >= 4 or total_defects > 10:
return "Critical - Immediate attention required"
elif max_severity >= 3 or total_defects > 5:
return "Poor - Repairs needed"
elif max_severity >= 2 or total_defects > 2:
return "Fair - Monitor and plan repairs"
else:
return "Good - Minor issues only"
def _generate_recommendations(self, defects: List[DetectedDefect],
element_type: str) -> List[str]:
"""Generate repair recommendations"""
recommendations = []
# Group defects by type
defect_groups = {}
for d in defects:
t = d.defect_type
if t not in defect_groups:
defect_groups[t] = []
defect_groups[t].append(d)
# Generate recommendations by defect type
for defect_type, group in defect_groups.items():
max_severity = max(d.severity for d in group)
if defect_type == DefectType.CRACK:
if max_severity in [SeverityLevel.CRITICAL, SeverityLevel.SEVERE]:
recommendations.append(
f"Structural engineer assessment required for {len(group)} crack(s). "
f"Consider epoxy injection or structural repair."
)
else:
recommendations.append(
f"Seal {len(group)} minor crack(s) with appropriate sealant."
)
elif defect_type == DefectType.SPALLING:
recommendations.append(
f"Remove loose concrete and apply repair mortar to {len(group)} spalling area(s). "
f"Check reinforcement for corrosion."
)
elif defect_type == DefectType.CORROSION:
recommendations.append(
f"Treat {len(group)} corrosion area(s). Clean rust, apply rust converter, "
f"and protective coating."
)
elif defect_type == DefectType.HONEYCOMBING:
recommendations.append(
f"Fill {len(group)} honeycomb area(s) with non-shrink grout. "
f"Investigate concrete placement procedures."
)
elif defect_type == DefectType.EFFLORESCENCE:
recommendations.append(
f"Clean efflorescence from {len(group)} area(s). "
f"Investigate and address moisture source."
)
if not recommendations:
recommendations.append("Continue regular inspection schedule.")
return recommendations
def export_inspection_report(self, inspection_id: str,
output_path: str) -> str:
"""Export inspection report to Excel"""
inspection = next(
(i for i in self.inspections if i.inspection_id == inspection_id),
None
)
if not inspection:
raise ValueError(f"Inspection {inspection_id} not found")
with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
# Summary
summary = pd.DataFrame([{
'Inspection ID': inspection.inspection_id,
'Date': inspection.timestamp.strftime('%Y-%m-%d %H:%M'),
'Location': inspection.location,
'Element Type': inspection.element_type,
'Overall Condition': inspection.overall_condition,
'Total Defects': len(inspection.defects),
'Image': inspection.image_path
}])
summary.to_excel(writer, sheet_name='Summary', index=False)
# Defects
if inspection.defects:
defect_data = [{
'Defect ID': d.defect_id,
'Type': d.defect_type.value,
'Severity': d.severity.value,
'Confidence': f"{d.confidence:.1%}",
'Location (x,y)': f"({d.bounding_box.x1}, {d.bounding_box.y1})",
'Size (w×h)': f"{d.bounding_box.width}×{d.bounding_box.height}",
'Dimensions (mm)': d.dimensions_mm if d.dimensions_mm else 'N/A'
} for d in inspection.defects]
pd.DataFrame(defect_data).to_excel(writer, sheet_name='Defects', index=False)
# Recommendations
rec_data = [{'#': i+1, 'Recommendation': r}
for i, r in enumerate(inspection.recommended_actions)]
pd.DataFrame(rec_data).to_excel(writer, sheet_name='Recommendations', index=False)
return output_path
def get_defect_statistics(self, start_date: datetime = None,
end_date: datetime = None) -> Dict:
"""Get defect statistics across inspections"""
filtered = self.inspections
if start_date:
filtered = [i for i in filtered if i.timestamp >= start_date]
if end_date:
filtered = [i for i in filtered if i.timestamp <= end_date]
all_defects = []
for inspection in filtered:
all_defects.extend(inspection.defects)
if not all_defects:
return {'message': 'No defects found in period'}
# Statistics
by_type = {}
by_severity = {}
for d in all_defects:
t = d.defect_type.value
s = d.severity.value
by_type[t] = by_type.get(t, 0) + 1
by_severity[s] = by_severity.get(s, 0) + 1
return {
'period': {
'start': start_date.isoformat() if start_date else 'all',
'end': end_date.isoformat() if end_date else 'all'
},
'total_inspections': len(filtered),
'total_defects': len(all_defects),
'by_type': by_type,
'by_severity': by_severity,
'avg_defects_per_inspection': len(all_defects) / len(filtered) if filtered else 0
}
模型训练
import torch
from torch.utils.data import Dataset, DataLoader
from torchvision import transforms
import os
from PIL import Image
class DefectDataset(Dataset):
"""Dataset for training defect detection model"""
def __init__(self, root_dir: str, annotations_file: str, transform=None):
self.root_dir = root_dir
self.annotations = self._load_annotations(annotations_file)
self.transform = transform or transforms.Compose([
transforms.Resize((800, 800)),
transforms.ToTensor()
])
def _load_annotations(self, path: str) -> List[Dict]:
"""Load COCO-format annotations"""
import json
with open(path, 'r') as f:
data = json.load(f)
return data['annotations']
def __len__(self):
return len(self.annotations)
def __getitem__(self, idx):
ann = self.annotations[idx]
image_path = os.path.join(self.root_dir, ann['image_file'])
image = Image.open(image_path).convert('RGB')
if self.transform:
image = self.transform(image)
# Prepare target
boxes = torch.tensor(ann['boxes'], dtype=torch.float32)
labels = torch.tensor(ann['labels'], dtype=torch.int64)
target = {
'boxes': boxes,
'labels': labels
}
return image, target
def train_defect_model(train_dataset: DefectDataset,
val_dataset: DefectDataset,
num_epochs: int = 10,
batch_size: int = 4,
learning_rate: float = 0.005):
"""Train defect detection model"""
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
# Initialize model
model = fasterrcnn_resnet50_fpn(pretrained=True)
num_classes = 8 # 7 defect types + background
in_features = model.roi_heads.box_predictor.cls_score.in_features
model.roi_heads.box_predictor = FastRCNNPredictor(in_features, num_classes)
model.to(device)
# Data loaders
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True,
collate_fn=lambda x: tuple(zip(*x)))
val_loader = DataLoader(val_dataset, batch_size=batch_size, shuffle=False,
collate_fn=lambda x: tuple(zip(*x)))
# Optimizer
optimizer = torch.optim.SGD(model.parameters(), lr=learning_rate,
momentum=0.9, weight_decay=0.0005)
# Training loop
for epoch in range(num_epochs):
model.train()
total_loss = 0
for images, targets in train_loader:
images = [img.to(device) for img in images]
targets = [{k: v.to(device) for k, v in t.items()} for t in targets]
loss_dict = model(images, targets)
losses = sum(loss for loss in loss_dict.values())
optimizer.zero_grad()
losses.backward()
optimizer.step()
total_loss += losses.item()
avg_loss = total_loss / len(train_loader)
print(f"Epoch {epoch+1}/{num_epochs}, Loss: {avg_loss:.4f}")
return model
快速参考
| 缺陷类型 | 检测方法 | 典型严重程度 |
|---|---|---|
| 裂缝 | 边缘检测 + CNN | 因宽度而异 |
| 剥落 | 目标检测 | 中等至严重 |
| 腐蚀 | 颜色 + 纹理分析 | 中等至严重 |
| 蜂窝 | 目标检测 | 严重 |
| 泛碱 | 颜色分析 | 轻微至中等 |
ACI 224R 裂缝宽度指南
| 宽度(mm) | 状况 | 暴露环境 |
|---|---|---|
| < 0.1 | 可接受 | 任意 |
| 0.1 - 0.2 | 可接受 | 干燥 |
| 0.2 - 0.4 | 建议修复 | 潮湿 |
| > 0.4 | 必须修复 | 任意 |
| > 1.0 | 需关注结构安全 | 任意 |
资源
- PyTorch:https://pytorch.org
- OpenCV:https://opencv.org
- ACI 224R:混凝土裂缝控制
- DDC 网站:https://datadrivenconstruction.io
后续步骤
- 参阅
progress-monitoring-cv,了解施工进度分析 - 参阅
safety-compliance-checker,了解安全缺陷集成 - 参阅
bim-validation-pipeline,了解基于模型的质量控制
评论
支持 Markdown