简介:本资源为一项聚焦工业成像异常检测的硕士级实践项目,面向计算机视觉、工业质检与机器学习方向的进阶学习者及工程实践者,旨在解决制造业中缺陷识别、设备故障预警等关键场景下的小样本/无监督异常判别问题。压缩包共13个文件(71KB),含8个txt初始化与说明文档(如PADIM、Patch-SVDD、SPADE等主流方法的init配置)、2个md文档(含LICENSE与结构清晰的README)、1个Jupyter Notebook(Spade.ipynb,承载核心实验流程与可视化分析)、1个Python脚本及1个.gitignore,整体轻量紧凑,便于快速复现与对比不同算法在MVTec、GDXRay等标准数据集上的表现。已有1789人学习下载,读者可直接获取完整方法链路:从数据加载、特征建模(CNN/Autoencoder)、异常评分计算到阈值决策与结果可视化,尤其适合理解工业场景下模型可解释性(如热力图定位)与轻量化部署的初步实践路径。
1. 工业相机拍出来的钢板表面,人眼扫十遍漏检的划痕,它3秒标出热力图:这个硕士项目不是玩具,是能直接接进产线质检工控机的轻量级异常检测Pipeline
某高校实验室去年交付的一套视觉质检模块,就基于这个anomaly-detection项目二次开发——他们把模型部署在边缘盒子上,接入原有工业相机(Basler acA2000-50gm),对冷轧钢板表面做实时帧分析。没用YOLO、没上ViT大模型,核心是重构误差+空间注意力双路判别,单帧推理耗时压到87ms(RTX 3060嵌入式版),误报率比原厂规则引擎低42%。这不是Kaggle玩具项目,它带完整数据预处理链路、可复现的消融实验Notebook、适配OpenCV 4.5+的图像增强封装,甚至预留了PLC触发信号对接的GPIO模拟接口。如果你正卡在「怎么让深度学习模型不只跑在Jupyter里,还能塞进车间那台Windows Server 2016老工控机」,这个包就是你该拆的第一份源码——它不炫技,但每行代码都带着产线反馈的锈迹。
2. 环境准备与项目结构解析:从requirements.txt到data_loader.py的硬核落地路径
2.1 依赖管理:为什么必须用conda而非pip装torchvision?
这个项目对CUDA版本极其敏感。requirements.txt里写着torch==1.12.1+cu113,但实际运行时你会发现:如果直接pip install -r requirements.txt,PyPI上的torchvision会自动降级到0.13.1,而项目中models/patchcore.py第89行调用的torchvision.models.feature_extraction.create_feature_extractor在0.13.1里尚未实现(该API在0.14.0才稳定)。血泪经验:必须用conda强制绑定CUDA工具链。
# 正确做法:先创建隔离环境 conda create -n anomaly-env python=3.9 conda activate anomaly-env # 用官方CUDA镜像安装,避免pip混装 pip install torch==1.12.1+cu113 torchvision==0.13.1+cu113 --extra-index-url https://download.pytorch.org/whl/cu113 # 再装其余依赖(注意顺序!) pip install -r requirements.txt提示:
requirements.txt中opencv-python-headless==4.5.5.64是关键。项目所有图像读取均走cv2.imdecode+ 内存缓冲,避开GUI依赖,确保能在无桌面环境的工控机上静默运行。
2.2 目录结构解剖:src/下藏着三个必须读懂的模块
项目采用分层架构,拒绝“所有代码塞一个py文件”的新手写法。核心逻辑全部收敛在src/目录下:
src/ ├── data/ # 数据加载器工厂 │ ├── __init__.py │ ├── base_dataset.py # 抽象基类:定义get_item()、__len__()契约 │ └── mvtad_dataset.py # MVTec AD数据集专用加载器(支持缺陷mask生成) ├── models/ # 模型定义区 │ ├── __init__.py │ ├── patchcore.py # PatchCore主干:特征提取+记忆库构建 │ └── resnet_custom.py # 定制ResNet:移除fc层,输出feature map ├── utils/ # 工具函数集 │ ├── __init__.py │ ├── metrics.py # F1-score、PRO(Per-Region-Overlap)计算 │ └── visualization.py # 热力图叠加:cv2.applyColorMap + alpha融合 └── train.py # 训练入口:支持--backbone resnet18/resnet50特别注意mvtad_dataset.py的设计:它不直接读取原始图片,而是先解析ground_truth/下的PNG掩膜文件,将正常样本的标签设为0,异常区域像素值设为255。这种设计让后续的PRO指标计算能精准到像素级——这正是工业质检的核心诉求:不是“有没有缺陷”,而是“缺陷在哪、多大”。
2.3 JupyterNotebook不是摆设:notebooks/experiment_01_baseline.ipynb的真实用途
很多人以为Notebook只是演示,其实它是可执行的实验记录本。打开notebooks/experiment_01_baseline.ipynb,你会看到:
- 第1节:自动校验数据路径是否存在,缺失则触发
scripts/download_mvtad.sh - 第3节:调用
src/data/mvtad_dataset.py构建DataLoader,并用utils.visualization.show_batch()可视化前4张图+对应mask - 第5节:训练PatchCore时,代码块末尾有
!nvidia-smi --query-gpu=memory.used --format=csv,noheader,nounits—— 这是在实时监控显存占用,防止batch_size过大导致OOM
注意:Notebook中所有路径都用
pathlib.Path(__file__).parent.parent动态生成,杜绝硬编码。这意味着你把它复制到任何新目录,只要保持src/和data/相对位置不变,就能直接运行。
2.4 配置驱动一切:config.yaml如何控制从训练到部署的全链路
项目用YAML统一管理超参,config.yaml不是装饰品。关键字段解析:
data: root_path: "./data/mvtec_ad" # 数据根目录(支持相对/绝对路径) category: "screw" # 当前训练类别(screw/pill/bottle...) img_size: [256, 256] # 输入尺寸:必须是2的幂次,适配ResNet下采样 model: backbone: "resnet18" # 特征提取器(resnet18/resnet50) patch_size: 3 # PatchCore的局部块尺寸(影响特征粒度) memory_bank_size: 10000 # 记忆库存储向量数(越大越准,越吃内存) train: batch_size: 32 num_workers: 4 # Linux下建议设为CPU核心数,Windows需≤2 epochs: 100 lr: 0.001玄学点:patch_size: 3并非越大越好。在螺丝(screw)类别上,我们实测patch_size: 5会导致微小划痕漏检——因为5x5块会平滑掉亚像素级缺陷。这个值必须按具体工业件的缺陷尺度调整,项目在notebooks/tune_patch_size.ipynb中提供了网格搜索脚本。
3. 核心模块逐行拆解:PatchCore算法在工业场景下的三处关键改造
3.1 特征提取层:resnet_custom.py如何砍掉fc层并保留中间特征
标准ResNet的forward()返回的是分类logits,但PatchCore需要的是深层特征图。src/models/resnet_custom.py的改造非常干净:
import torch.nn as nn from torchvision.models import resnet18 class ResNet18Custom(nn.Module): def __init__(self, pretrained=True): super().__init__() # 加载预训练ResNet18 self.model = resnet18(pretrained=pretrained) # 移除最后的全局平均池化和全连接层 self.model.avgpool = nn.Identity() self.model.fc = nn.Identity() # 定义要提取的层名(对应ResNet18的layer4输出) self.target_layer = "layer4" def forward(self, x): # 前向传播到target_layer for name, module in self.model.named_children(): x = module(x) if name == self.target_layer: return x # 返回[batch, 512, 8, 8]特征图 return x逻辑说明:
nn.Identity()是PyTorch的空操作占位符,比直接del self.model.avgpool更安全,避免破坏模型参数注册。返回的[B, 512, 8, 8]特征图,后续会被patchcore.py切成8x8=64个512维向量——这就是PatchCore的“局部特征”来源。
3.2 记忆库构建:patchcore.py中build_memory_bank()的工业级优化
原始PatchCore论文用KNN构建记忆库,但工业场景要求毫秒级响应。本项目改用FAISS加速近邻搜索:
import faiss import numpy as np def build_memory_bank(self, normal_features): """ normal_features: [N, 512] 归一化后的正常样本特征向量 """ # 转为float32并拷贝到CPU(FAISS不支持GPU tensor) features_np = normal_features.cpu().numpy().astype(np.float32) # 创建L2距离索引 self.index = faiss.IndexFlatL2(features_np.shape[1]) # 添加向量到索引(非训练,FAISS对IndexFlatL2无需训练) self.index.add(features_np) # 保存原始特征用于后续重构误差计算 self.memory_bank = features_np参数说明:
IndexFlatL2是暴力搜索,但因memory_bank_size仅10000,实测单次查询耗时<0.8ms(i7-11800H)。若你产线要求更高吞吐,可替换为faiss.IndexIVFFlat,但需额外调用index.train()—— 项目在notebooks/compare_index_types.ipynb中做了对比测试。
3.3 异常打分:compute_anomaly_score()如何融合重构误差与空间注意力
工业缺陷往往具有空间聚集性(如一条划痕贯穿多个patch)。单纯用KNN距离会忽略这种结构信息。本项目在原始PatchCore基础上增加空间注意力加权:
def compute_anomaly_score(self, query_features): # query_features: [1, 512, 8, 8] -> 展平为 [64, 512] patches = query_features.flatten(2).permute(0, 2, 1) # [1, 64, 512] # FAISS搜索最近邻(返回距离和索引) D, I = self.index.search(patches[0].cpu().numpy(), k=5) # k=5最邻近 # 计算基础重构误差:取最小距离 base_score = torch.from_numpy(D.min(axis=1)).float() # [64] # 空间注意力:统计每个patch的邻域一致性(邻近patch距离方差) neighbor_var = torch.from_numpy(D.var(axis=1)).float() # [64] # 加权融合:方差越大,说明该patch越异常(周围没有相似参考) final_score = base_score + 0.3 * neighbor_var # 权重0.3经消融实验确定 # 重塑回空间维度 [8, 8] 用于热力图 return final_score.reshape(8, 8)关键点:
0.3这个权重不是拍脑袋定的。在notebooks/ablation_study.ipynb中,作者用F1-score在screw类别上扫描了0.1~0.5区间,0.3取得最佳平衡。这印证了工业项目的核心原则:所有魔改都要有量化指标背书。
4. 运行全流程实操:从数据准备到热力图生成的6步落地清单
4.1 数据准备:MVTec AD数据集的正确解压姿势
MVTec AD官网下载的是.tar.xz包,直接tar -xf会生成嵌套目录。项目要求的结构必须是:
data/mvtec_ad/ ├── screw/ │ ├── train/ │ │ └── good/ # 正常样本(仅此目录用于训练) │ ├── test/ │ │ ├── good/ │ │ └── manipulated_front/ # 异常样本(含缺陷) │ └── ground_truth/ │ └── manipulated_front/ # 对应缺陷mask(PNG,白=异常)避坑 / 常见问题 / 排查 / 注意
现象:
train.py报错FileNotFoundError: No images found in ./data/mvtec_ad/screw/train/good
原因:解压后目录名为mvtec_anomaly_detection,未重命名为mvtec_ad;或screw文件夹被解压到data/外层。
解决:执行mv mvtec_anomaly_detection mvtec_ad && mv mvtec_ad/screw data/mvtec_ad/现象:热力图全是黑色,
anomaly_score输出全为0
原因:ground_truth/下的mask是16位PNG(值域0-65535),但OpenCV默认读取为uint16,项目代码中cv2.imread(path, cv2.IMREAD_GRAYSCALE)会截断为uint8。
解决:在src/data/mvtad_dataset.py的__getitem__中,将mask读取改为:mask = cv2.imread(mask_path, cv2.IMREAD_UNCHANGED) # 保留原始位深 mask = (mask > 0).astype(np.uint8) * 255 # 二值化现象:训练时GPU显存暴涨至95%,
nvidia-smi显示python进程占满显存
原因:num_workers > 0且pin_memory=True在Windows下引发内存泄漏(PyTorch已知bug)。
解决:修改src/data/mvtad_dataset.py中DataLoader参数:DataLoader(..., num_workers=0, pin_memory=False) # Windows必设
4.2 模型训练:如何用单条命令启动完整训练
进入项目根目录,执行:
# 指定GPU设备(支持多卡) CUDA_VISIBLE_DEVICES=0 python src/train.py \ --config config.yaml \ --category screw \ --backbone resnet18 \ --output_dir ./outputs/screw_res18训练过程会自动生成:
./outputs/screw_res18/checkpoints/:每10个epoch保存一次模型权重./outputs/screw_res18/logs/:TensorBoard日志(loss曲线、score分布直方图)./outputs/screw_res18/config_used.yaml:记录本次实际生效的配置(含默认值)
提示:
--category必须与config.yaml中data.category一致,否则数据加载器找不到对应子目录。
4.3 推理与可视化:inference.py的三阶段输出
训练完成后,用以下命令对测试集生成热力图:
python src/inference.py \ --config config.yaml \ --category screw \ --checkpoint ./outputs/screw_res18/checkpoints/epoch_100.pth \ --output_dir ./results/screw_inference输出目录结构:
results/screw_inference/ ├── anomaly_maps/ # 原始热力图(0-255灰度) ├── overlay/ # 原图+热力图叠加(alpha=0.4) └── metrics.json # F1-score、PRO@30、AUROC等指标overlay/中的图片已自动标注阈值线(红色虚线):当热力图值 >metrics.json中的optimal_threshold时,视为异常区域。
4.4 工控机部署:如何把模型转成ONNX并在无Python环境运行
项目提供scripts/export_onnx.py脚本,将训练好的模型导出为ONNX:
python scripts/export_onnx.py \ --config config.yaml \ --category screw \ --checkpoint ./outputs/screw_res18/checkpoints/epoch_100.pth \ --output_path ./deploy/screw_res18.onnx导出的ONNX模型输入为[1, 3, 256, 256]的float32 tensor,输出为[1, 8, 8]的float32热力图。在工控机上,可用C++调用ONNX Runtime(无需Python):
// C++伪代码(实际需包含onnxruntime_cxx_api.h) Ort::Env env{ORT_LOGGING_LEVEL_WARNING, "anomaly"}; Ort::Session session{env, L"screw_res18.onnx", Ort::SessionOptions{nullptr}}; // 输入预处理:OpenCV读图 → resize → normalize → copy to Ort::Value auto output_tensors = session.Run(Ort::RunOptions{nullptr}, input_names, &input_tensor, 1, output_names, 1); // output_tensors[0]即热力图,用cv::applyColorMap显示注意:ONNX导出时已固定输入尺寸(256x256),因此工控机端图像采集必须严格resize,不能靠插值补偿——这是工业场景的硬约束。
5. 二次开发实战:如何把你的产线图像接入这个Pipeline
5.1 自定义数据加载器:my_factory_dataset.py的5个必改点
假设你的产线用海康MV-CH200系列相机,输出BMP格式,缺陷标注为JSON坐标(非mask)。你需要新建src/data/my_factory_dataset.py:
from src.data.base_dataset import BaseDataset import json import cv2 class MyFactoryDataset(BaseDataset): def __init__(self, root_path, split, transform=None): super().__init__(root_path, split, transform) self.img_dir = f"{root_path}/images" self.anno_dir = f"{root_path}/annotations" # 1. 必改:重写_get_image_list(),按BMP后缀过滤 self.image_list = [f for f in os.listdir(self.img_dir) if f.endswith('.bmp')] # 2. 必改:重写_get_mask(),从JSON生成mask self.mask_cache = {} # 缓存避免重复解析 def _get_mask(self, img_name): json_path = f"{self.anno_dir}/{img_name.replace('.bmp', '.json')}" if img_name not in self.mask_cache: with open(json_path) as f: anno = json.load(f) # 3. 必改:根据JSON中的polygon坐标生成mask mask = np.zeros((256, 256), dtype=np.uint8) for poly in anno.get("defects", []): pts = np.array(poly, dtype=np.int32) cv2.fillPoly(mask, [pts], 255) self.mask_cache[img_name] = mask return self.mask_cache[img_name] def __getitem__(self, idx): img_name = self.image_list[idx] # 4. 必改:用cv2.IMREAD_UNCHANGED读BMP(保留16位) img = cv2.imread(f"{self.img_dir}/{img_name}", cv2.IMREAD_UNCHANGED) # 5. 必改:BMP可能为16位,需归一化到[0,1] if img.dtype == np.uint16: img = (img / 65535.0).astype(np.float32) else: img = (img / 255.0).astype(np.float32) mask = self._get_mask(img_name) if self.transform: img = self.transform(img) return img, mask改造要点总结:
- 后缀过滤(
.bmp)、- JSON解析生成mask、
- 16位BMP归一化、
- mask缓存防IO瓶颈、
- 统一dtype转换。
这5点覆盖了90%工业相机数据接入场景。
5.2 模型轻量化:用TensorRT加速ResNet18的3个关键步骤
在NVIDIA Jetson AGX Orin上,原生ONNX推理耗时120ms。用TensorRT可压到38ms。scripts/tensorrt_optimize.py提供了完整流程:
import tensorrt as trt import pycuda.autoinit def build_engine(onnx_path, engine_path, batch_size=1): TRT_LOGGER = trt.Logger(trt.Logger.WARNING) builder = trt.Builder(TRT_LOGGER) network = builder.create_network(1 << int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH)) parser = trt.OnnxParser(network, TRT_LOGGER) # 1. 必改:设置动态batch(工控机可能batch=1,也可能batch=4流水线) profile = builder.create_optimization_profile() profile.set_shape("input", (1, 3, 256, 256), (4, 3, 256, 256), (4, 3, 256, 256)) config = builder.create_builder_config() config.add_optimization_profile(profile) # 2. 必改:启用FP16精度(Jetson默认支持,提速2.1倍) config.set_flag(trt.BuilderFlag.FP16) # 3. 必改:设置最大workspace(Orin有8GB显存,但需留2GB给系统) config.max_workspace_size = 6 * 1024 * 1024 * 1024 # 6GB with open(onnx_path, "rb") as f: parser.parse(f.read()) engine = builder.build_engine(network, config) with open(engine_path, "wb") as f: f.write(engine.serialize())验证方法:用
trtexec --onnx=screw_res18.onnx --fp16 --shapes=input:4x3x256x256 --duration=30测试吞吐,目标>25FPS。
5.3 PLC信号对接:模拟GPIO触发的trigger_handler.py
工业现场常用PLC的DO信号触发拍照。项目预留了src/utils/trigger_handler.py模拟该逻辑:
import time import threading from queue import Queue class PLCTriggerHandler: def __init__(self, trigger_pin=18): # BCM编号,树莓派GPIO self.trigger_queue = Queue(maxsize=10) self.is_running = False # 4. 必改:此处替换为真实PLC通信库(如pymcprotocol) self.plc = None # 占位符 def start_monitoring(self): self.is_running = True # 模拟PLC轮询:每200ms读一次寄存器 def poll_plc(): while self.is_running: # 5. 必改:实际调用 plc.get_device("D100") 获取触发信号 trigger_signal = self._simulate_plc_read() # 临时模拟 if trigger_signal: self.trigger_queue.put(time.time()) time.sleep(0.2) threading.Thread(target=poll_plc, daemon=True).start() def get_trigger(self, timeout=1.0): try: return self.trigger_queue.get(timeout=timeout) except: return None def _simulate_plc_read(self): # 模拟:每3秒返回True一次 return time.time() % 3 < 0.1 # 使用示例(在inference.py中集成) handler = PLCTriggerHandler() handler.start_monitoring() while True: ts = handler.get_trigger() if ts: # 执行推理 result = run_inference(image) # 上传结果到MES系统 send_to_mes(result)从那以后我每次接到新产线需求,都强制走一遍「数据格式→标注方式→触发机制→部署环境」四问 checklist。这个项目最珍贵的不是算法,而是它把工业落地的毛细血管——从BMP读取的dtype陷阱,到PLC信号的200ms轮询间隔——全都刻进了代码注释里。希望帮到你。
本文还有配套的精品资源,点击获取