企业官网建设流程全解析

在安防监控、消防救援、夜间巡检等场景中，传统可见光摄像头受光线条件限制明显，而热成像技术通过检测物体发出的红外辐射，能够在完全黑暗、烟雾、雾霾等恶劣环境下清晰成像。结合YOLOv8这一当前最先进的实时目标检测算法，可以构建出高精度的热成像人员识别系统，为24小时不间断安防监控提供可靠技术支撑。

本文将完整介绍基于YOLOv8的热成像人员检测系统从环境搭建、数据准备、模型训练到完整系统部署的全流程。无论你是刚接触深度学习的新手，还是有一定计算机视觉基础的开发者，都能通过本文掌握完整的项目实施方法。文章包含详细的代码示例、配置文件、常见问题解决方案，所有内容均可直接复用到实际项目中。

1. 项目背景与技术选型

1.1 热成像技术优势与应用场景

热成像相机通过检测物体表面的红外辐射强度来生成图像，不同于传统相机依赖可见光反射。这种技术具有几大核心优势：

• 全天候工作能力：完全不受光照条件影响，可在漆黑环境中正常检测
• 穿透能力：能够穿透烟雾、薄雾等遮挡物，在火灾救援等场景中尤为重要
• 隐私保护：只能检测人体轮廓，无法识别具体面部特征，适用于需要保护个人隐私的监控场景
• 温度检测：可同时获取目标的表面温度信息，在疫情防控、工业检测中有额外价值

典型应用场景包括：夜间安防监控、森林防火人员搜救、化工厂危险区域监控、疫情防控体温筛查、电力设备巡检等。

1.2 YOLOv8算法优势

YOLOv8是Ultralytics公司在2023年发布的最新版本目标检测算法，相比前代具有显著改进：

• 更高的检测精度：采用新的骨干网络和检测头设计，在COCO数据集上达到更高mAP
• 更快的推理速度：优化了网络结构和训练策略，在相同硬件条件下速度提升明显
• 更友好的使用体验：提供了更简洁的API接口和更完善的文档支持
• 多任务支持：除了目标检测，还支持实例分割、姿态估计等任务

对于热成像人员检测这种需要实时处理的应用场景，YOLOv8的效率和精度优势尤为明显。

1.3 系统架构概述

完整的系统包含以下几个核心模块：

1. 数据采集模块：热成像相机或热成像视频流输入
2. 预处理模块：图像增强、尺寸标准化等预处理操作
3. 推理检测模块：YOLOv8模型加载和推理计算
4. 后处理模块：检测结果解析、非极大值抑制等
5. 可视化界面：检测结果实时显示和报警功能

2. 环境准备与依赖安装

2.1 系统环境要求

推荐使用以下环境配置，其他版本可能需要适当调整：

• 操作系统：Ubuntu 20.04/22.04或Windows 10/11
• Python版本：3.8-3.10（推荐3.9）
• 深度学习框架：PyTorch 1.12.0及以上
• CUDA版本：11.3-11.7（GPU加速可选）

2.2 核心依赖安装

创建并激活Python虚拟环境：

# 创建虚拟环境 python -m venv yolov8_thermal # 激活环境（Linux/Mac） source yolov8_thermal/bin/activate # 激活环境（Windows） yolov8_thermal\Scripts\activate # 升级pip版本 pip install --upgrade pip

安装PyTorch（根据CUDA版本选择）：

# 如果使用GPU（CUDA 11.7） pip install torch==1.13.1+cu117 torchvision==0.14.1+cu117 torchaudio==0.13.1 --extra-index-url https://download.pytorch.org/whl/cu117 # 如果仅使用CPU pip install torch==1.13.1+cpu torchvision==0.14.1+cpu torchaudio==0.13.1 --extra-index-url https://download.pytorch.org/whl/cpu

安装YOLOv8和其他依赖：

# 安装Ultralytics YOLOv8 pip install ultralytics # 安装图像处理和相关库 pip install opencv-python pillow numpy pandas matplotlib seaborn # 安装界面开发库 pip install PyQt5 pyqt5-tools # 安装其他工具库 pip install scipy tqdm tensorboard

2.3 环境验证

创建验证脚本检查环境是否正确安装：

# environment_check.py import torch import cv2 from ultralytics import YOLO import PyQt5 print(f"PyTorch版本: {torch.__version__}") print(f"CUDA可用: {torch.cuda.is_available()}") if torch.cuda.is_available(): print(f"GPU设备: {torch.cuda.get_device_name(0)}") print(f"OpenCV版本: {cv2.__version__}") print(f"YOLOv8版本: {YOLO.__version__}") print("环境验证通过！")

运行验证脚本确保所有依赖正确安装。

3. 热成像数据集准备与处理

3.1 热成像数据特点

热成像数据与普通RGB图像有显著差异：

• 单通道灰度图像：通常为16位或8位灰度图
• 温度值映射：像素值对应实际温度或相对温度
• 对比度较低：需要适当的图像增强处理
• 标注标准：标注框需要适应热成像中的人体轮廓特征

3.2 数据收集与标注

热成像人员检测数据集可以通过以下方式获取：

1. 公开数据集：FLIR ADAS等公开热成像数据集
2. 自采集数据：使用热成像相机实际采集
3. 数据合成：将可见光数据转换为热成像风格

创建数据标注规范文件：

# data.yaml path: /path/to/thermal_dataset # 数据集根目录 train: images/train # 训练图像路径 val: images/val # 验证图像路径 test: images/test # 测试图像路径 # 类别定义 nc: 1 # 类别数量（仅检测人员） names: ['person'] # 类别名称 # 自动下载权重 download: https://ultralytics.com/static/assets/datasets/coco8.zip

3.3 数据预处理增强

针对热成像特点设计数据增强策略：

# data_augmentation.py import cv2 import numpy as np from PIL import Image, ImageEnhance class ThermalDataAugmentation: def __init__(self): self.augmentations = [ self.adjust_contrast, self.adjust_brightness, self.gaussian_blur, self.thermal_noise ] def adjust_contrast(self, image, factor=1.5): """增强热成像对比度""" enhancer = ImageEnhance.Contrast(Image.fromarray(image)) return np.array(enhancer.enhance(factor)) def adjust_brightness(self, image, factor=1.2): """调整亮度""" enhancer = ImageEnhance.Brightness(Image.fromarray(image)) return np.array(enhancer.enhance(factor)) def gaussian_blur(self, image, kernel_size=3): """高斯模糊模拟热成像模糊效应""" return cv2.GaussianBlur(image, (kernel_size, kernel_size), 0) def thermal_noise(self, image, intensity=0.01): """添加热噪声模拟真实热成像""" noise = np.random.normal(0, intensity, image.shape) noisy_image = image + noise * 255 return np.clip(noisy_image, 0, 255).astype(np.uint8) def apply_random_augmentation(self, image): """随机应用一种增强方法""" aug_func = np.random.choice(self.augmentations) return aug_func(image) # 使用示例 augmentor = ThermalDataAugmentation() augmented_image = augmentor.apply_random_augmentation(original_image)

4. YOLOv8模型训练与优化

4.1 模型选择与配置

YOLOv8提供多种规模的预训练模型，根据实际需求选择：

# model_config.py from ultralytics import YOLO class YOLOv8Config: # 模型规模选择 MODEL_SIZES = { 'nano': 'yolov8n.pt', # 最轻量，速度最快 'small': 'yolov8s.pt', # 平衡型 'medium': 'yolov8m.pt', # 精度较高 'large': 'yolov8l.pt', # 高精度 'xlarge': 'yolov8x.pt' # 最高精度 } # 训练参数配置 TRAIN_CONFIG = { 'epochs': 100, 'imgsz': 640, 'batch': 16, 'workers': 4, 'lr0': 0.01, # 初始学习率 'lrf': 0.01, # 最终学习率 'momentum': 0.937, # 动量 'weight_decay': 0.0005, # 权重衰减 'warmup_epochs': 3.0, # 热身轮数 'warmup_momentum': 0.8,# 热身动量 'box': 7.5, # 框损失权重 'cls': 0.5, # 分类损失权重 'dfl': 1.5 # DFL损失权重 } def select_model(size='medium'): """根据需求选择模型""" model_name = YOLOv8Config.MODEL_SIZES.get(size, 'yolov8m.pt') return YOLO(model_name)

4.2 训练流程实现

完整的模型训练流程：

# train_thermal_yolov8.py import os from ultralytics import YOLO import yaml class ThermalYOLOv8Trainer: def __init__(self, data_config, model_size='medium'): self.data_config = data_config self.model = YOLO(f'yolov8{model_size[0]}.pt') # 加载预训练模型 def setup_training_config(self): """设置训练配置""" config = { 'data': self.data_config, 'epochs': 100, 'imgsz': 640, 'batch': 16, 'patience': 10, # 早停耐心值 'save': True, 'exist_ok': False, # 不覆盖现有实验 'pretrained': True, 'optimizer': 'auto', 'verbose': True, 'seed': 42, 'deterministic': True } return config def start_training(self): """开始训练""" config = self.setup_training_config() # 开始训练 results = self.model.train( data=config['data'], epochs=config['epochs'], imgsz=config['imgsz'], batch=config['batch'], patience=config['patience'], save=config['save'], pretrained=config['pretrained'], optimizer=config['optimizer'], verbose=config['verbose'], seed=config['seed'], deterministic=config['deterministic'] ) return results def evaluate_model(self): """模型评估""" metrics = self.model.val() print(f"mAP50-95: {metrics.box.map:.4f}") print(f"mAP50: {metrics.box.map50:.4f}") print(f"Precision: {metrics.box.precision:.4f}") print(f"Recall: {metrics.box.recall:.4f}") return metrics # 使用示例 if __name__ == "__main__": trainer = ThermalYOLOv8Trainer('thermal_data.yaml', 'medium') training_results = trainer.start_training() evaluation_metrics = trainer.evaluate_model()

4.3 模型导出与优化

训练完成后导出为不同格式以适应各种部署环境：

# model_export.py from ultralytics import YOLO class ModelExporter: def __init__(self, model_path): self.model = YOLO(model_path) def export_onnx(self, output_path, simplify=True): """导出为ONNX格式""" success = self.model.export( format='onnx', imgsz=640, simplify=simplify, opset=12 ) return success def export_tensorrt(self, output_path): """导出为TensorRT引擎""" success = self.model.export( format='engine', imgsz=640, device=0 # GPU设备 ) return success def export_openvino(self, output_path): """导出为OpenVINO格式""" success = self.model.export( format='openvino', imgsz=640 ) return success def export_all_formats(self, base_name): """批量导出所有格式""" formats = ['onnx', 'engine', 'openvino'] results = {} for fmt in formats: output_path = f"{base_name}.{fmt}" try: success = self.model.export(format=fmt, imgsz=640) results[fmt] = success print(f"{fmt.upper()}导出{'成功' if success else '失败'}") except Exception as e: results[fmt] = False print(f"{fmt.upper()}导出失败: {e}") return results # 使用示例 exporter = ModelExporter('runs/detect/train/weights/best.pt') exporter.export_all_formats('thermal_person_detector')

5. 系统界面开发与集成

5.1 PyQt5界面设计

使用PyQt5开发用户友好的检测系统界面：

# thermal_ui.py import sys import cv2 from PyQt5.QtWidgets import (QApplication, QMainWindow, QVBoxLayout, QHBoxLayout, QPushButton, QLabel, QWidget, QFileDialog, QComboBox, QSlider, QSpinBox, QGroupBox, QTextEdit, QProgressBar) from PyQt5.QtCore import QTimer, Qt, pyqtSignal, QThread from PyQt5.QtGui import QImage, QPixmap from ultralytics import YOLO import numpy as np class DetectionThread(QThread): """检测线程，避免界面卡顿""" frame_processed = pyqtSignal(np.ndarray, list) def __init__(self, model_path, camera_index=0): super().__init__() self.model = YOLO(model_path) self.camera_index = camera_index self.is_running = False self.cap = None def run(self): """线程主循环""" self.cap = cv2.VideoCapture(self.camera_index) self.is_running = True while self.is_running: ret, frame = self.cap.read() if not ret: break # 执行检测 results = self.model(frame, imgsz=640, conf=0.5) # 绘制检测结果 annotated_frame = results[0].plot() # 获取检测信息 detections = [] for box in results[0].boxes: detections.append({ 'class': results[0].names[int(box.cls)], 'confidence': float(box.conf), 'bbox': box.xyxy[0].tolist() }) # 发送处理后的帧和检测结果 self.frame_processed.emit(annotated_frame, detections) if self.cap: self.cap.release() def stop(self): """停止检测""" self.is_running = False class ThermalDetectionUI(QMainWindow): """热成像人员检测主界面""" def __init__(self): super().__init__() self.detection_thread = None self.init_ui() def init_ui(self): """初始化界面""" self.setWindowTitle("YOLOv8热成像人员检测系统") self.setGeometry(100, 100, 1200, 800) # 中央窗口部件 central_widget = QWidget() self.setCentralWidget(central_widget) # 主布局 main_layout = QHBoxLayout() central_widget.setLayout(main_layout) # 左侧视频显示区域 left_layout = QVBoxLayout() # 视频显示标签 self.video_label = QLabel() self.video_label.setMinimumSize(800, 600) self.video_label.setStyleSheet("border: 2px solid gray;") self.video_label.setAlignment(Qt.AlignCenter) self.video_label.setText("视频显示区域") left_layout.addWidget(self.video_label) # 控制按钮 control_layout = QHBoxLayout() self.start_btn = QPushButton("开始检测") self.stop_btn = QPushButton("停止检测") self.load_video_btn = QPushButton("加载视频") self.snapshot_btn = QPushButton("截图") self.start_btn.clicked.connect(self.start_detection) self.stop_btn.clicked.connect(self.stop_detection) self.load_video_btn.clicked.connect(self.load_video_file) self.snapshot_btn.clicked.connect(self.take_snapshot) control_layout.addWidget(self.start_btn) control_layout.addWidget(self.stop_btn) control_layout.addWidget(self.load_video_btn) control_layout.addWidget(self.snapshot_btn) left_layout.addLayout(control_layout) # 右侧信息面板 right_layout = QVBoxLayout() # 检测设置组 settings_group = QGroupBox("检测设置") settings_layout = QVBoxLayout() # 置信度阈值设置 conf_layout = QHBoxLayout() conf_layout.addWidget(QLabel("置信度阈值:")) self.conf_slider = QSlider(Qt.Horizontal) self.conf_slider.setRange(10, 90) self.conf_slider.setValue(50) self.conf_value = QLabel("0.5") conf_layout.addWidget(self.conf_slider) conf_layout.addWidget(self.conf_value) settings_layout.addLayout(conf_layout) # 模型选择 model_layout = QHBoxLayout() model_layout.addWidget(QLabel("模型选择:")) self.model_combo = QComboBox() self.model_combo.addItems(["YOLOv8n", "YOLOv8s", "YOLOv8m", "YOLOv8l"]) self.model_combo.setCurrentIndex(2) model_layout.addWidget(self.model_combo) settings_layout.addLayout(model_layout) settings_group.setLayout(settings_layout) right_layout.addWidget(settings_group) # 检测结果组 results_group = QGroupBox("检测结果") results_layout = QVBoxLayout() self.results_text = QTextEdit() self.results_text.setReadOnly(True) results_layout.addWidget(self.results_text) # 统计信息 stats_layout = QHBoxLayout() stats_layout.addWidget(QLabel("检测人数:")) self.person_count = QLabel("0") stats_layout.addWidget(self.person_count) stats_layout.addWidget(QLabel("平均置信度:")) self.avg_confidence = QLabel("0.00") stats_layout.addWidget(self.avg_confidence) results_layout.addLayout(stats_layout) results_group.setLayout(results_layout) right_layout.addWidget(results_group) # 进度条 self.progress_bar = QProgressBar() right_layout.addWidget(self.progress_bar) # 组合左右布局 main_layout.addLayout(left_layout, 3) main_layout.addLayout(right_layout, 1) def start_detection(self): """开始检测""" if self.detection_thread and self.detection_thread.isRunning(): return model_map = {"YOLOv8n": "yolov8n", "YOLOv8s": "yolov8s", "YOLOv8m": "yolov8m", "YOLOv8l": "yolov8l"} selected_model = model_map[self.model_combo.currentText()] self.detection_thread = DetectionThread(f"{selected_model}.pt") self.detection_thread.frame_processed.connect(self.update_frame) self.detection_thread.start() def stop_detection(self): """停止检测""" if self.detection_thread: self.detection_thread.stop() self.detection_thread.wait() def update_frame(self, frame, detections): """更新视频帧显示""" # 转换OpenCV BGR到Qt RGB rgb_image = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) h, w, ch = rgb_image.shape bytes_per_line = ch * w qt_image = QImage(rgb_image.data, w, h, bytes_per_line, QImage.Format_RGB888) # 缩放图像适应显示区域 pixmap = QPixmap.fromImage(qt_image) scaled_pixmap = pixmap.scaled(self.video_label.size(), Qt.KeepAspectRatio) self.video_label.setPixmap(scaled_pixmap) # 更新检测结果 self.update_detection_results(detections) def update_detection_results(self, detections): """更新检测结果显示""" person_detections = [d for d in detections if d['class'] == 'person'] self.person_count.setText(str(len(person_detections))) if person_detections: avg_conf = sum(d['confidence'] for d in person_detections) / len(person_detections) self.avg_confidence.setText(f"{avg_conf:.3f}") else: self.avg_confidence.setText("0.00") # 更新文本显示 results_text = f"检测到 {len(person_detections)} 个人员\n\n" for i, detection in enumerate(person_detections, 1): results_text += f"人员 {i}: 置信度 {detection['confidence']:.3f}\n" self.results_text.setText(results_text) def load_video_file(self): """加载视频文件""" file_path, _ = QFileDialog.getOpenFileName( self, "选择视频文件", "", "视频文件 (*.mp4 *.avi *.mov)") if file_path: # 实现视频文件加载逻辑 pass def take_snapshot(self): """截图功能""" # 实现截图保存逻辑 pass def main(): app = QApplication(sys.argv) window = ThermalDetectionUI() window.show() sys.exit(app.exec_()) if __name__ == "__main__": main()

5.2 实时视频流处理

实现热成像相机视频流接入和处理：

# video_stream.py import cv2 import threading import time from queue import Queue class ThermalCameraStream: """热成像相机流处理类""" def __init__(self, source=0, model_path='yolov8m.pt'): self.source = source self.model = YOLO(model_path) self.cap = cv2.VideoCapture(source) self.frame_queue = Queue(maxsize=10) self.detection_queue = Queue(maxsize=10) self.is_running = False self.threads = [] # 相机参数设置（根据具体热成像相机调整） self.setup_camera() def setup_camera(self): """设置热成像相机参数""" if self.cap.isOpened(): # 设置分辨率（根据相机支持调整） self.cap.set(cv2.CAP_PROP_FRAME_WIDTH, 640) self.cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 480) # 设置帧率 self.cap.set(cv2.CAP_PROP_FPS, 30) def start_stream(self): """开始视频流处理""" self.is_running = True # 创建帧捕获线程 capture_thread = threading.Thread(target=self.capture_frames) capture_thread.daemon = True capture_thread.start() self.threads.append(capture_thread) # 创建检测线程 detection_thread = threading.Thread(target=self.process_detections) detection_thread.daemon = True detection_thread.start() self.threads.append(detection_thread) def capture_frames(self): """捕获视频帧""" while self.is_running: ret, frame = self.cap.read() if ret: # 如果队列已满，丢弃最旧的帧 if self.frame_queue.full(): try: self.frame_queue.get_nowait() except: pass self.frame_queue.put(frame) time.sleep(0.01) # 控制捕获频率 def process_detections(self): """处理目标检测""" while self.is_running: if not self.frame_queue.empty(): frame = self.frame_queue.get() # 执行YOLOv8检测 results = self.model(frame, imgsz=640, conf=0.5) # 获取检测结果 detections = [] for box in results[0].boxes: detections.append({ 'class': results[0].names[int(box.cls)], 'confidence': float(box.conf), 'bbox': box.xyxy[0].tolist() }) # 绘制检测结果 annotated_frame = results[0].plot() # 放入结果队列 if self.detection_queue.full(): try: self.detection_queue.get_nowait() except: pass self.detection_queue.put((annotated_frame, detections)) def get_latest_detection(self): """获取最新的检测结果""" if not self.detection_queue.empty(): return self.detection_queue.get() return None, [] def stop_stream(self): """停止视频流""" self.is_running = False if self.cap: self.cap.release() # 等待线程结束 for thread in self.threads: thread.join(timeout=1.0) # 使用示例 def demo_thermal_stream(): stream = ThermalCameraStream(source=0) # 0为默认相机 stream.start_stream() try: while True: frame, detections = stream.get_latest_detection() if frame is not None: cv2.imshow('Thermal Detection', frame) # 显示检测信息 person_count = len([d for d in detections if d['class'] == 'person']) print(f"检测到 {person_count} 个人员") if cv2.waitKey(1) & 0xFF == ord('q'): break finally: stream.stop_stream() cv2.destroyAllWindows()

6. 系统部署与性能优化

6.1 模型量化与加速

针对边缘设备部署进行模型优化：

# model_optimization.py import torch from ultralytics import YOLO import onnxruntime as ort class ModelOptimizer: def __init__(self, model_path): self.model = YOLO(model_path) def quantize_model(self, output_path): """模型量化减小尺寸""" # 加载模型 model = torch.jit.load(self.model) # 动态量化 quantized_model = torch.quantization.quantize_dynamic( model, {torch.nn.Linear}, dtype=torch.qint8 ) # 保存量化模型 torch.jit.save(quantized_model, output_path) return output_path def optimize_for_inference(self): """推理优化""" # 设置模型为评估模式 self.model.eval() # 启用推理优化 with torch.no_grad(): # 使用torch.jit.trace优化 example_input = torch.rand(1, 3, 640, 640) traced_model = torch.jit.trace(self.model, example_input) return traced_model def benchmark_performance(self, test_data, iterations=100): """性能基准测试""" model = self.optimize_for_inference() # Warmup for _ in range(10): _ = model(test_data) # 基准测试 start_time = torch.cuda.Event(enable_timing=True) end_time = torch.cuda.Event(enable_timing=True) start_time.record() for _ in range(iterations): _ = model(test_data) end_time.record() torch.cuda.synchronize() inference_time = start_time.elapsed_time(end_time) / iterations print(f"平均推理时间: {inference_time:.2f}ms") print(f"帧率: {1000/inference_time:.2f}FPS") return inference_time class ONNXInference: """ONNX运行时推理优化""" def __init__(self, onnx_path): self.session = ort.InferenceSession(onnx_path) self.providers = ['CUDAExecutionProvider', 'CPUExecutionProvider'] def optimize_session(self): """优化推理会话""" # 设置优化选项 options = self.session.get_provider_options() # CUDA优化设置 if 'CUDAExecutionProvider' in self.providers: options['CUDAExecutionProvider'] = { 'arena_extend_strategy': 'kNextPowerOfTwo', 'cudnn_conv_algo_search': 'EXHAUSTIVE', 'do_copy_in_default_stream': True, } return self.session def inference(self, input_data): """执行推理""" input_name = self.session.get_inputs()[0].name output_name = self.session.get_outputs()[0].name results = self.session.run([output_name], {input_name: input_data}) return results

6.2 多线程处理架构

设计高效的多线程处理架构：

# multi_thread_processing.py import threading import queue import time from concurrent.futures import ThreadPoolExecutor import cv2 class PipelineProcessor: """流水线处理器""" def __init__(self, model_path, num_workers=4): self.model_path = model_path self.num_workers = num_workers self.input_queue = queue.Queue(maxsize=20) self.output_queue = queue.Queue(maxsize=20) self.is_running = False self.workers = [] def start_processing(self): """启动处理流水线""" self.is_running = True # 创建工作线程池 with ThreadPoolExecutor(max_workers=self.num_workers) as executor: while self.is_running: try: # 从输入队列获取帧 frame_data = self.input_queue.get(timeout=1.0) if frame_data is None: # 停止信号 break # 提交检测任务 future = executor.submit(self.process_frame, frame_data) future.add_done_callback(self.on_processing_done) except queue.Empty: continue def process_frame(self, frame_data): """处理单帧图像""" frame, frame_id = frame_data # 执行YOLOv8检测 from ultralytics import YOLO model = YOLO(self.model_path) results = model(frame, imgsz=640, conf=0.5) # 提取检测结果 detections = [] for box in results[0].boxes: detections.append({ 'class': results[0].names[int(box.cls)], 'confidence': float(box.conf), 'bbox': box.xyxy[0].tolist() }) annotated_frame = results[0].plot() return annotated_frame, detections, frame_id def on_processing_done(self, future): """处理完成回调""" try: result = future.result() self.output_queue.put(result) except Exception as e: print(f"处理失败: {e}") def add_frame(self, frame, frame_id): """添加帧到处理队列""" if self.input_queue.full(): # 队列满时丢弃最旧的帧 try: self.input_queue.get_nowait() except queue.Empty: pass self.input_queue.put((frame, frame_id)) def get_result(self): """获取处理结果""" try: return self.output_queue.get_nowait() except queue.Empty: return None def stop_processing(self): """停止处理""" self.is_running = False self.input_queue.put(None) # 发送停止信号 # 使用示例 class ThermalDetectionSystem: """完整的检测系统""" def __init__(self, model_path, camera_source=0): self.camera_source = camera_source self.processor = PipelineProcessor(model_path) self.cap = cv2.VideoCapture(camera_source) self.is_running = False def start_system(self): """启动系统""" self.is_running = True self.processor.start_processing() frame_id = 0 while self.is_running: ret, frame = self.cap.read() if ret: self.processor.add_frame(frame, frame_id) frame_id += 1 # 获取并显示结果 result = self.processor.get_result() if result: annotated_frame, detections, fid = result cv2.imshow('Thermal Detection', annotated_frame) # 显示检测统计 person_count = len([d for d in detections if d['class'] == 'person']) print(f"帧 {fid}: 检测到 {person_count} 人") if cv2.waitKey(1) & 0xFF == ord('q'): break self.stop_system() def stop_system(self): """停止系统""" self.is_running = False self.processor.stop_processing() if self.cap: self.cap.release() cv2.destroyAllWindows()

7. 常见问题与解决方案

7.1 环境配置问题

问题1：CUDA out of memory错误

解决方案：

# 方法1：减小批处理大小 model.train(batch=8) # 从16减小到8 # 方法2：使用梯度累积 model.train(batch=8, accumulate=2) # 等效批大小16 # 方法3：启用混合精度训练 model.train(amp=True) #