哈密市网站建设_网站建设公司_CSS_seo优化

摘要

本文详细介绍了基于YOLO系列算法（YOLOv5/YOLOv6/YOLOv7/YOLOv8）的火焰与烟雾检测系统。该系统集成了先进的深度学习目标检测技术，结合用户友好的UI界面，能够实时检测视频和图像中的火焰与烟雾。本文将从算法原理、系统架构、数据集构建、模型训练、性能优化和完整代码实现等多个角度进行全面阐述

目录

摘要

一、引言

1.1 研究背景与意义

1.2 YOLO算法的发展历程

二、系统架构设计

2.1 总体架构

2.2 技术栈

三、数据集构建与预处理

3.1 参考数据集

3.1.1 公开数据集推荐

3.1.2 数据集结构

3.2 数据预处理流程

四、YOLO模型详解与实现

4.1 YOLOv8模型架构

4.2 多模型集成策略

五、模型训练与优化

5.1 训练配置

5.2 训练脚本

六、UI界面设计与实现

6.1 PyQt5主界面

6.2 检测线程实现

七、系统集成与部署

7.1 主程序入口

7.2 Docker部署配置

八、性能评估与优化

8.1 评估指标计算

8.2 模型优化技术

九、实际应用案例

9.1 工业厂房火灾预警系统

9.2 森林火灾监测无人机

一、引言

1.1 研究背景与意义

火灾是威胁人类生命财产安全的重大灾害之一，早期火灾检测对于减少损失至关重要。传统的火灾检测方法主要基于传感器技术，但存在响应延迟、易受环境干扰等问题。基于计算机视觉的火焰与烟雾检测技术具有非接触、实时性强、覆盖范围广等优势，成为火灾预警领域的研究热点。

1.2 YOLO算法的发展历程

YOLO（You Only Look Once）系列算法作为单阶段目标检测的代表，自2016年首次提出以来，经历了多个版本的迭代优化：

• YOLOv1-v3：奠定了YOLO系列的基础架构

• YOLOv4：引入Bag of Freebies和Bag of Specials策略

• YOLOv5：采用PyTorch框架，简化部署流程

• YOLOv6：针对工业应用优化

• YOLOv7：提出扩展高效层聚合网络

• YOLOv8：最新版本，在精度和速度间取得更好平衡

二、系统架构设计

2.1 总体架构

本系统采用模块化设计，主要包括以下组件：

text

火焰烟雾检测系统架构： 1. 数据预处理模块 2. 深度学习模型模块 3. 推理引擎模块 4. UI界面模块 5. 报警与日志模块

2.2 技术栈

• 深度学习框架：PyTorch

• 前端界面：PyQt5/Tkinter

• 数据处理：OpenCV, PIL, NumPy

• 模型部署：ONNX, TensorRT（可选）

• 开发环境：Python 3.8+

三、数据集构建与预处理

3.1 参考数据集

3.1.1 公开数据集推荐

1. Fire Detection Dataset

   • 来源：Kaggle

   • 内容：包含火灾图像和正常场景图像

   • 规模：约10,000张图像

   • 标注：边界框标注

2. SMOKE Dataset

   • 来源：IEEE DataPort

   • 内容：烟雾检测专用数据集

   • 规模：5,000+张图像

   • 特点：包含不同环境下的烟雾场景

3. Custom Dataset（自制数据集）

   • 采集来源：监控视频、网络图像、模拟实验

   • 标注工具：LabelImg/Roboflow

   • 数据增强策略：翻转、旋转、色彩调整、混合增强

3.1.2 数据集结构

python

dataset/ ├── images/ │ ├── train/ │ ├── val/ │ └── test/ ├── labels/ │ ├── train/ │ ├── val/ │ └── test/ ├── data.yaml └── classes.txt

3.2 数据预处理流程

python

import cv2 import numpy as np import albumentations as A from albumentations.pytorch import ToTensorV2 class FireSmokeDataAugmentation: def __init__(self, img_size=640): self.train_transform = A.Compose([ A.Resize(height=img_size, width=img_size), A.HorizontalFlip(p=0.5), A.VerticalFlip(p=0.1), A.RandomBrightnessContrast(p=0.2), A.RandomGamma(p=0.2), A.Blur(blur_limit=3, p=0.1), A.MedianBlur(blur_limit=3, p=0.1), A.ToGray(p=0.1), A.CLAHE(p=0.1), A.RandomRotate90(p=0.2), A.Cutout(num_holes=8, max_h_size=32, max_w_size=32, p=0.5), A.Normalize( mean=[0, 0, 0], std=[1, 1, 1], max_pixel_value=255.0 ), ToTensorV2() ], bbox_params=A.BboxParams( format='yolo', label_fields=['class_labels'] )) def apply_augmentation(self, image, bboxes, class_labels): transformed = self.train_transform( image=image, bboxes=bboxes, class_labels=class_labels ) return transformed['image'], transformed['bboxes'], transformed['class_labels']

四、YOLO模型详解与实现

4.1 YOLOv8模型架构

python

import torch import torch.nn as nn from ultralytics import YOLO class EnhancedYOLOv8: def __init__(self, model_size='m', num_classes=2): """ 初始化增强版YOLOv8模型 Args: model_size: 模型尺寸 (n, s, m, l, x) num_classes: 类别数（火焰、烟雾） """ self.model_size = model_size self.num_classes = num_classes self.model = self.build_model() def build_model(self): """构建YOLOv8模型""" # 加载预训练模型 model = YOLO(f'yolov8{self.model_size}.pt') # 修改分类头以适应火焰烟雾检测 if self.num_classes != 80: # 默认COCO 80类 # 获取模型结构 model.model.nc = self.num_classes # 修改检测头的输出通道 for m in model.model.modules(): if isinstance(m, nn.Conv2d): if m.weight.shape[0] == 255: # 默认输出通道 # 重新初始化检测头 m.out_channels = (self.num_classes + 5) * 3 m.weight = nn.Parameter( torch.randn(m.weight.shape[0], m.weight.shape[1], m.weight.shape[2], m.weight.shape[3]) ) return model def add_attention_module(self): """添加注意力机制模块""" class CBAM(nn.Module): """Convolutional Block Attention Module""" def __init__(self, channels, reduction=16): super(CBAM, self).__init__() # 通道注意力 self.avg_pool = nn.AdaptiveAvgPool2d(1) self.max_pool = nn.AdaptiveMaxPool2d(1) self.fc = nn.Sequential( nn.Linear(channels, channels // reduction, bias=False), nn.ReLU(inplace=True), nn.Linear(channels // reduction, channels, bias=False) ) self.sigmoid = nn.Sigmoid() # 空间注意力 self.conv = nn.Conv2d(2, 1, kernel_size=7, padding=3, bias=False) def forward(self, x): # 通道注意力 b, c, _, _ = x.size() avg_out = self.fc(self.avg_pool(x).view(b, c)) max_out = self.fc(self.max_pool(x).view(b, c)) channel_attention = self.sigmoid(avg_out + max_out).view(b, c, 1, 1) x = x * channel_attention # 空间注意力 avg_out = torch.mean(x, dim=1, keepdim=True) max_out, _ = torch.max(x, dim=1, keepdim=True) spatial_attention = torch.cat([avg_out, max_out], dim=1) spatial_attention = self.sigmoid(self.conv(spatial_attention)) return x * spatial_attention # 在骨干网络关键位置添加CBAM for i, layer in enumerate(self.model.model.model): if isinstance(layer, nn.Conv2d) and layer.out_channels >= 128: # 在特定层后添加注意力模块 cbam = CBAM(layer.out_channels) self.model.model.model[i] = nn.Sequential(layer, cbam)

4.2 多模型集成策略

python

class ModelEnsemble: """多模型集成类""" def __init__(self, model_paths, ensemble_method='weighted'): """ Args: model_paths: 模型路径列表 ensemble_method: 集成方法 ('weighted', 'nms', 'wbf') """ self.models = [] for path in model_paths: if 'yolov5' in path: model = torch.hub.load('ultralytics/yolov5', 'custom', path=path) elif 'yolov8' in path: model = YOLO(path) self.models.append(model) self.ensemble_method = ensemble_method def weighted_box_fusion(self, detections, weights=None): """加权框融合""" if weights is None: weights = [1/len(detections)] * len(detections) fused_boxes = [] fused_scores = [] fused_labels = [] # 实现WBF算法 for i, det in enumerate(detections): if len(det) > 0: for box in det: # 计算融合权重 weight = weights[i] # 框融合逻辑 # ... 具体实现 return fused_boxes, fused_scores, fused_labels

五、模型训练与优化

5.1 训练配置

yaml

# config/training_config.yaml train: # 数据配置 data: "dataset/data.yaml" epochs: 300 batch_size: 16 imgsz: 640 workers: 8 # 优化器配置 optimizer: "AdamW" lr0: 0.001 lrf: 0.01 momentum: 0.937 weight_decay: 0.0005 warmup_epochs: 3 warmup_momentum: 0.8 # 数据增强 hsv_h: 0.015 hsv_s: 0.7 hsv_v: 0.4 degrees: 0.0 translate: 0.1 scale: 0.5 shear: 0.0 perspective: 0.0 flipud: 0.0 fliplr: 0.5 mosaic: 1.0 mixup: 0.0 # 模型保存 save_period: 10 pretrained: true freeze: 10 val: val_period: 1 save_json: false test: test_period: 10

5.2 训练脚本

python

import yaml import torch from torch.utils.data import DataLoader import wandb from tqdm import tqdm class FireSmokeTrainer: def __init__(self, config_path): with open(config_path, 'r') as f: self.config = yaml.safe_load(f) self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') self.setup_training() def setup_training(self): """设置训练环境""" # 初始化wandb wandb.init(project="fire-smoke-detection", config=self.config) # 创建数据加载器 self.train_loader, self.val_loader = self.create_dataloaders() # 初始化模型 self.model = EnhancedYOLOv8( model_size=self.config['model_size'], num_classes=2 ).model.to(self.device) # 设置优化器 self.optimizer = self.configure_optimizer() # 学习率调度器 self.scheduler = torch.optim.lr_scheduler.CosineAnnealingWarmRestarts( self.optimizer, T_0=10, T_mult=2 ) # 损失函数 self.criterion = self.configure_loss() def train_epoch(self, epoch): """训练单个epoch""" self.model.train() total_loss = 0 pbar = tqdm(self.train_loader, desc=f'Epoch {epoch}') for batch_idx, (images, targets) in enumerate(pbar): images = images.to(self.device) targets = targets.to(self.device) # 前向传播 self.optimizer.zero_grad() outputs = self.model(images) # 计算损失 loss = self.criterion(outputs, targets) # 反向传播 loss.backward() # 梯度裁剪 torch.nn.utils.clip_grad_norm_(self.model.parameters(), max_norm=10.0) self.optimizer.step() total_loss += loss.item() # 更新进度条 pbar.set_postfix({'loss': loss.item()}) # wandb记录 wandb.log({ 'train/loss': loss.item(), 'train/epoch': epoch, 'train/lr': self.optimizer.param_groups[0]['lr'] }) return total_loss / len(self.train_loader) def validate(self, epoch): """验证模型""" self.model.eval() val_loss = 0 metrics = { 'precision': 0, 'recall': 0, 'mAP@0.5': 0, 'mAP@0.5:0.95': 0 } with torch.no_grad(): for images, targets in tqdm(self.val_loader, desc='Validating'): images = images.to(self.device) targets = targets.to(self.device) outputs = self.model(images) loss = self.criterion(outputs, targets) val_loss += loss.item() # 计算评估指标 # ... 实现评估逻辑 return val_loss / len(self.val_loader), metrics def train(self): """主训练循环""" best_map = 0 for epoch in range(self.config['train']['epochs']): # 训练阶段 train_loss = self.train_epoch(epoch) # 验证阶段 if epoch % self.config['val']['val_period'] == 0: val_loss, metrics = self.validate(epoch) # 保存最佳模型 if metrics['mAP@0.5:0.95'] > best_map: best_map = metrics['mAP@0.5:0.95'] torch.save({ 'epoch': epoch, 'model_state_dict': self.model.state_dict(), 'optimizer_state_dict': self.optimizer.state_dict(), 'best_map': best_map, 'config': self.config }, f'best_model_epoch{epoch}.pth') # wandb记录验证指标 wandb.log({ 'val/loss': val_loss, 'val/mAP': metrics['mAP@0.5:0.95'], 'val/precision': metrics['precision'], 'val/recall': metrics['recall'] }) # 学习率调整 self.scheduler.step()

六、UI界面设计与实现

6.1 PyQt5主界面

python

import sys from PyQt5.QtWidgets import * from PyQt5.QtCore import * from PyQt5.QtGui import * import cv2 import numpy as np class FireSmokeDetectionUI(QMainWindow): def __init__(self): super().__init__() self.model = None self.detection_thread = None self.init_ui() self.load_model() def init_ui(self): """初始化用户界面""" self.setWindowTitle("火焰烟雾智能检测系统 v2.0") self.setGeometry(100, 100, 1400, 800) # 中央部件 central_widget = QWidget() self.setCentralWidget(central_widget) # 主布局 main_layout = QHBoxLayout() # 左侧视频显示区域 left_panel = QVBoxLayout() # 视频显示标签 self.video_label = QLabel() self.video_label.setMinimumSize(800, 600) self.video_label.setStyleSheet("border: 2px solid #ccc; background-color: black;") left_panel.addWidget(self.video_label) # 控制按钮区域 control_layout = QHBoxLayout() self.btn_open_camera = QPushButton("打开摄像头") self.btn_open_camera.clicked.connect(self.open_camera) self.btn_open_video = QPushButton("打开视频文件") self.btn_open_video.clicked.connect(self.open_video_file) self.btn_open_image = QPushButton("打开图片") self.btn_open_image.clicked.connect(self.open_image_file) self.btn_stop = QPushButton("停止检测") self.btn_stop.clicked.connect(self.stop_detection) self.btn_stop.setEnabled(False) self.btn_settings = QPushButton("系统设置") self.btn_settings.clicked.connect(self.open_settings) control_layout.addWidget(self.btn_open_camera) control_layout.addWidget(self.btn_open_video) control_layout.addWidget(self.btn_open_image) control_layout.addWidget(self.btn_stop) control_layout.addWidget(self.btn_settings) control_layout.addStretch() left_panel.addLayout(control_layout) # 右侧信息面板 right_panel = QVBoxLayout() # 统计信息区域 stats_group = QGroupBox("检测统计") stats_layout = QVBoxLayout() self.fire_count_label = QLabel("火焰数量: 0") self.smoke_count_label = QLabel("烟雾数量: 0") self.confidence_label = QLabel("平均置信度: 0.00") self.fps_label = QLabel("FPS: 0") stats_layout.addWidget(self.fire_count_label) stats_layout.addWidget(self.smoke_count_label) stats_layout.addWidget(self.confidence_label) stats_layout.addWidget(self.fps_label) # 报警阈值设置 threshold_layout = QHBoxLayout() threshold_layout.addWidget(QLabel("报警阈值:")) self.threshold_slider = QSlider(Qt.Horizontal) self.threshold_slider.setRange(0, 100) self.threshold_slider.setValue(50) self.threshold_slider.valueChanged.connect(self.update_threshold) self.threshold_label = QLabel("0.50") threshold_layout.addWidget(self.threshold_slider) threshold_layout.addWidget(self.threshold_label) stats_layout.addLayout(threshold_layout) stats_group.setLayout(stats_layout) right_panel.addWidget(stats_group) # 日志区域 log_group = QGroupBox("系统日志") log_layout = QVBoxLayout() self.log_text = QTextEdit() self.log_text.setReadOnly(True) self.log_text.setMaximumHeight(200) log_layout.addWidget(self.log_text) log_group.setLayout(log_layout) right_panel.addWidget(log_group) # 模型选择区域 model_group = QGroupBox("模型配置") model_layout = QVBoxLayout() self.model_combo = QComboBox() self.model_combo.addItems(["YOLOv8n", "YOLOv8s", "YOLOv8m", "YOLOv8l", "YOLOv8x"]) self.model_combo.currentTextChanged.connect(self.change_model) model_layout.addWidget(QLabel("选择模型:")) model_layout.addWidget(self.model_combo) # 检测模式 mode_layout = QHBoxLayout() self.realtime_radio = QRadioButton("实时检测") self.realtime_radio.setChecked(True) self.accuracy_radio = QRadioButton("精确模式") mode_layout.addWidget(self.realtime_radio) mode_layout.addWidget(self.accuracy_radio) model_layout.addLayout(mode_layout) model_group.setLayout(model_layout) right_panel.addWidget(model_group) # 报警设置 alarm_group = QGroupBox("报警设置") alarm_layout = QVBoxLayout() self.alarm_checkbox = QCheckBox("启用声音报警") self.alarm_checkbox.setChecked(True) self.email_checkbox = QCheckBox("启用邮件通知") alarm_layout.addWidget(self.alarm_checkbox) alarm_layout.addWidget(self.email_checkbox) # 报警历史按钮 self.btn_alarm_history = QPushButton("查看报警历史") self.btn_alarm_history.clicked.connect(self.show_alarm_history) alarm_layout.addWidget(self.btn_alarm_history) alarm_group.setLayout(alarm_layout) right_panel.addWidget(alarm_group) # 添加到主布局 main_layout.addLayout(left_panel, 7) main_layout.addLayout(right_panel, 3) central_widget.setLayout(main_layout) # 状态栏 self.statusBar().showMessage("系统就绪") # 初始化计时器 self.timer = QTimer() self.timer.timeout.connect(self.update_frame) # 报警系统 self.alarm_system = AlarmSystem() def update_threshold(self, value): """更新检测阈值""" threshold = value / 100.0 self.threshold_label.setText(f"{threshold:.2f}") if self.model: self.model.conf = threshold def log_message(self, message): """记录日志信息""" timestamp = QDateTime.currentDateTime().toString("hh:mm:ss") self.log_text.append(f"[{timestamp}] {message}") # 自动滚动到底部 scrollbar = self.log_text.verticalScrollBar() scrollbar.setValue(scrollbar.maximum())

6.2 检测线程实现

python

class DetectionThread(QThread): """检测线程""" update_signal = pyqtSignal(np.ndarray, list, float) stats_signal = pyqtSignal(dict) alarm_signal = pyqtSignal(str, str) def __init__(self, model, source=0): super().__init__() self.model = model self.source = source self.running = True self.confidence_threshold = 0.5 self.cap = None def run(self): """线程主函数""" if isinstance(self.source, str): self.cap = cv2.VideoCapture(self.source) else: self.cap = cv2.VideoCapture(self.source) fps = 0 frame_count = 0 start_time = time.time() while self.running and self.cap.isOpened(): ret, frame = self.cap.read() if not ret: break # 调整帧大小 frame = cv2.resize(frame, (800, 600)) # 进行检测 results = self.model(frame, conf=self.confidence_threshold) # 解析结果 detections = [] fire_count = 0 smoke_count = 0 total_confidence = 0 for result in results: boxes = result.boxes if boxes is not None: for box in boxes: x1, y1, x2, y2 = box.xyxy[0].cpu().numpy() confidence = box.conf[0].cpu().numpy() class_id = int(box.cls[0].cpu().numpy()) # 类别映射 class_name = "火焰" if class_id == 0 else "烟雾" # 统计 if class_id == 0: fire_count += 1 else: smoke_count += 1 total_confidence += confidence # 保存检测结果 detections.append({ 'bbox': [x1, y1, x2, y2], 'confidence': confidence, 'class': class_name, 'class_id': class_id }) # 计算平均置信度 avg_confidence = total_confidence / len(detections) if detections else 0 # 计算FPS frame_count += 1 if frame_count % 30 == 0: fps = 30 / (time.time() - start_time) start_time = time.time() # 绘制检测框 annotated_frame = self.draw_detections(frame, detections) # 发送信号 self.update_signal.emit(annotated_frame, detections, fps) # 发送统计信息 stats = { 'fire_count': fire_count, 'smoke_count': smoke_count, 'avg_confidence': avg_confidence, 'fps': fps } self.stats_signal.emit(stats) # 检查是否需要报警 if fire_count > 0 or smoke_count > 0: alarm_msg = f"检测到{fire_count}处火焰，{smoke_count}处烟雾" self.alarm_signal.emit("warning", alarm_msg) # 控制帧率 time.sleep(0.01) def draw_detections(self, frame, detections): """在帧上绘制检测结果""" for det in detections: x1, y1, x2, y2 = map(int, det['bbox']) confidence = det['confidence'] class_name = det['class'] # 根据类别选择颜色 if det['class_id'] == 0: # 火焰 color = (0, 0, 255) # 红色 else: # 烟雾 color = (128, 128, 128) # 灰色 # 绘制边界框 cv2.rectangle(frame, (x1, y1), (x2, y2), color, 2) # 绘制标签背景 label = f"{class_name}: {confidence:.2f}" (text_width, text_height), _ = cv2.getTextSize( label, cv2.FONT_HERSHEY_SIMPLEX, 0.5, 2 ) cv2.rectangle( frame, (x1, y1 - text_height - 10), (x1 + text_width, y1), color, -1 ) # 绘制标签文本 cv2.putText( frame, label, (x1, y1 - 5), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 255, 255), 2 ) # 绘制关键点（火焰检测时可添加） if det['class_id'] == 0: # 在火焰中心绘制闪烁点 center_x = (x1 + x2) // 2 center_y = (y1 + y2) // 2 radius = int(min(x2 - x1, y2 - y1) * 0.1) # 创建闪烁效果 alpha = (int(time.time() * 10) % 10) / 10.0 overlay = frame.copy() cv2.circle(overlay, (center_x, center_y), radius, (0, 255, 255), -1) cv2.addWeighted(overlay, alpha, frame, 1 - alpha, 0, frame) return frame

七、系统集成与部署

7.1 主程序入口

python

import argparse import sys import os def parse_args(): parser = argparse.ArgumentParser(description="火焰烟雾检测系统") parser.add_argument('--model', type=str, default='yolov8m', help='模型类型: yolov5, yolov6, yolov7, yolov8') parser.add_argument('--weights', type=str, default='best.pt', help='模型权重路径') parser.add_argument('--source', type=str, default='0', help='数据源: 0(摄像头), 视频文件, 图片文件') parser.add_argument('--conf', type=float, default=0.5, help='检测置信度阈值') parser.add_argument('--iou', type=float, default=0.45, help='IOU阈值') parser.add_argument('--device', type=str, default='cuda', help='运行设备: cuda, cpu') parser.add_argument('--gui', action='store_true', help='启用GUI界面') parser.add_argument('--save', action='store_true', help='保存检测结果') parser.add_argument('--save_dir', type=str, default='results', help='结果保存目录') return parser.parse_args() def main(): args = parse_args() # 设置环境 os.environ['KMP_DUPLICATE_LIB_OK'] = 'TRUE' if args.gui: # GUI模式 app = QApplication(sys.argv) window = FireSmokeDetectionUI() # 加载模型 if args.model.startswith('yolov8'): from ultralytics import YOLO model = YOLO(args.weights) elif args.model.startswith('yolov5'): import torch model = torch.hub.load('ultralytics/yolov5', 'custom', path=args.weights) window.model = model window.show() sys.exit(app.exec_()) else: # 命令行模式 detector = FireSmokeDetector( model_type=args.model, weights_path=args.weights, device=args.device ) detector.detect( source=args.source, conf_threshold=args.conf, iou_threshold=args.iou, save_results=args.save, save_dir=args.save_dir ) if __name__ == "__main__": main()

7.2 Docker部署配置

dockerfile

# Dockerfile FROM pytorch/pytorch:2.0.0-cuda11.7-cudnn8-runtime # 设置工作目录 WORKDIR /app # 安装系统依赖 RUN apt-get update && apt-get install -y \ libgl1-mesa-glx \ libglib2.0-0 \ libsm6 \ libxext6 \ libxrender-dev \ libgomp1 \ wget \ && rm -rf /var/lib/apt/lists/* # 复制项目文件 COPY requirements.txt . COPY . . # 安装Python依赖 RUN pip install --upgrade pip RUN pip install -r requirements.txt # 下载预训练模型 RUN wget https://github.com/ultralytics/assets/releases/download/v0.0.0/yolov8m.pt -O weights/yolov8m.pt # 暴露端口 EXPOSE 8080 # 启动命令 CMD ["python", "main.py", "--gui", "--device", "cuda"]

八、性能评估与优化

8.1 评估指标计算

python

class ModelEvaluator: """模型评估器""" def __init__(self, model, test_loader, device='cuda'): self.model = model self.test_loader = test_loader self.device = device self.metrics = {} def evaluate(self): """全面评估模型性能""" results = {} # 基础指标 results['basic'] = self.calculate_basic_metrics() # 速度指标 results['speed'] = self.measure_inference_speed() # 鲁棒性测试 results['robustness'] = self.test_robustness() # 内存使用 results['memory'] = self.measure_memory_usage() return results def calculate_basic_metrics(self): """计算基础评估指标""" self.model.eval() all_predictions = [] all_targets = [] with torch.no_grad(): for images, targets in tqdm(self.test_loader, desc="Evaluating"): images = images.to(self.device) outputs = self.model(images) # 处理预测结果 predictions = self.process_predictions(outputs) all_predictions.extend(predictions) all_targets.extend(targets.cpu().numpy()) # 计算mAP map_50 = self.calculate_map(all_predictions, all_targets, iou_threshold=0.5) map_95 = self.calculate_map(all_predictions, all_targets, iou_threshold=0.5:0.95) # 计算精度和召回率 precision, recall = self.calculate_precision_recall(all_predictions, all_targets) return { 'mAP@0.5': map_50, 'mAP@0.5:0.95': map_95, 'precision': precision, 'recall': recall, 'f1_score': 2 * (precision * recall) / (precision + recall + 1e-16) } def measure_inference_speed(self): """测量推理速度""" warmup_iterations = 10 test_iterations = 100 # 预热 dummy_input = torch.randn(1, 3, 640, 640).to(self.device) for _ in range(warmup_iterations): _ = self.model(dummy_input) # 测试推理时间 torch.cuda.synchronize() start_time = time.time() for _ in range(test_iterations): _ = self.model(dummy_input) torch.cuda.synchronize() end_time = time.time() avg_time = (end_time - start_time) / test_iterations fps = 1 / avg_time return { 'inference_time_ms': avg_time * 1000, 'fps': fps, 'batch_size': 1 }

8.2 模型优化技术

python

class ModelOptimizer: """模型优化器""" def __init__(self, model): self.model = model self.optimized_model = None def prune_model(self, pruning_rate=0.3): """模型剪枝""" import torch.nn.utils.prune as prune parameters_to_prune = [] for name, module in self.model.named_modules(): if isinstance(module, torch.nn.Conv2d): parameters_to_prune.append((module, 'weight')) prune.global_unstructured( parameters_to_prune, pruning_method=prune.L1Unstructured, amount=pruning_rate, ) # 移除剪枝掩码 for module, param_name in parameters_to_prune: prune.remove(module, param_name) return self.model def quantize_model(self): """模型量化""" # 动态量化 quantized_model = torch.quantization.quantize_dynamic( self.model, {torch.nn.Linear, torch.nn.Conv2d}, dtype=torch.qint8 ) return quantized_model def export_onnx(self, input_shape=(1, 3, 640, 640), output_path="model.onnx"): """导出ONNX模型""" dummy_input = torch.randn(*input_shape) torch.onnx.export( self.model, dummy_input, output_path, export_params=True, opset_version=11, do_constant_folding=True, input_names=['input'], output_names=['output'], dynamic_axes={ 'input': {0: 'batch_size'}, 'output': {0: 'batch_size'} } ) # 验证ONNX模型 import onnx onnx_model = onnx.load(output_path) onnx.checker.check_model(onnx_model) return output_path

九、实际应用案例

9.1 工业厂房火灾预警系统

python

class IndustrialFireMonitoring: """工业厂房火灾监控系统""" def __init__(self, camera_configs): """ Args: camera_configs: 摄像头配置列表 """ self.cameras = [] self.detectors = [] self.alarm_history = [] # 初始化多个摄像头 for config in camera_configs: camera = IndustrialCamera( rtsp_url=config['rtsp_url'], location=config['location'], area=config['area'] ) self.cameras.append(camera) # 为每个摄像头创建检测器 detector = FireSmokeDetector( model_type='yolov8m', weights_path='weights/best_industrial.pt' ) self.detectors.append(detector) # 初始化中央监控 self.monitor_center = MonitorCenter() def start_monitoring(self): """开始监控""" threads = [] for i, (camera, detector) in enumerate(zip(self.cameras, self.detectors)): thread = threading.Thread( target=self.monitor_camera, args=(camera, detector, i) ) thread.daemon = True thread.start() threads.append(thread) # 启动中央监控界面 self.monitor_center.show() def monitor_camera(self, camera, detector, camera_id): """监控单个摄像头""" while True: try: # 获取视频帧 frame = camera.get_frame() if frame is not None: # 检测火焰烟雾 results = detector.detect_frame(frame) # 分析结果 analysis = self.analyze_results(results, camera.location) # 如果发现火灾风险 if analysis['risk_level'] >= 2: # 中等风险及以上 self.trigger_alarm(analysis, camera_id) # 更新监控中心 self.monitor_center.update_camera_view( camera_id, frame, results, analysis ) except Exception as e: logger.error(f"摄像头{camera_id}监控错误: {str(e)}") time.sleep(0.033) # 约30FPS

9.2 森林火灾监测无人机

python

class ForestFireDrone: """森林火灾监测无人机系统""" def __init__(self, drone_id, model_path): self.drone_id = drone_id self.model = YOLO(model_path) self.gps_tracker = GPSTracker() self.thermal_camera = ThermalCamera() self.communication = SatelliteCommunication() def patrol_mission(self, waypoints): """巡逻任务""" for waypoint in waypoints: # 飞往航点 self.fly_to(waypoint) # 悬停检测 self.hover_and_scan(duration=30) # 分析检测结果 analysis = self.analyze_detections() # 如果发现火灾 if analysis['fire_detected']: self.emergency_protocol(analysis) def hover_and_scan(self, duration): """悬停并扫描""" start_time = time.time() detections_buffer = [] while time.time() - start_time < duration: # 获取可见光图像 visible_image = self.get_visible_image() # 获取热成像图像 thermal_image = self.get_thermal_image() # 融合检测 visible_detections = self.model(visible_image) thermal_detections = self.analyze_thermal(thermal_image) # 多模态融合 fused_detections = self.fuse_detections( visible_detections, thermal_detections ) detections_buffer.append(fused_detections) # 实时传输数据 self.transmit_data(visible_image, fused_detections) time.sleep(1) return detections_buffer