基于YOLO系列模型的牧场羊行为检测系统完整实现方案

在牧场智能化管理系统中，羊只行为检测是一个关键技术需求。传统的人工观察方式效率低下且容易遗漏重要行为信息。本文将详细介绍基于YOLOv5/v8/11/12/26系列模型结合PyQt和MySQL的牧场羊行为检测系统完整实现方案。

1. 项目背景与需求分析

1.1 牧场智能化管理需求

现代牧场管理面临着羊只行为监测的诸多挑战：

• 传统人工观察耗时耗力，无法实现24小时不间断监测
• 难以准确记录羊只的进食、饮水、休息等行为模式
• 疾病早期症状难以被发现，容易延误治疗时机
• 繁殖期行为监测对提高繁殖率至关重要

1.2 技术方案选型理由

选择YOLO系列模型的原因：

• YOLOv5：成熟稳定，社区支持完善，适合快速部署
• YOLOv8：精度和速度平衡，新增功能丰富
• YOLOv11/v12：最新架构，在精度上有显著提升
• YOLOv26：针对边缘设备优化的轻量级版本

PyQt作为GUI框架的优势：

• 跨平台支持，可在Windows、Linux系统运行
• 界面美观，支持自定义控件开发
• 与Python生态完美集成

MySQL数据库的作用：

• 存储检测结果和历史数据
• 支持多用户并发访问
• 提供数据分析和报表功能

2. 环境搭建与依赖配置

2.1 系统环境要求

# 操作系统要求 - Ubuntu 18.04+ 或 Windows 10+ - Python 3.8+ - CUDA 11.3+ (GPU加速) - cuDNN 8.2+ # 硬件建议配置 - GPU: NVIDIA GTX 1660 以上 - 内存: 8GB+ - 存储: 500GB+ SSD

2.2 Python环境配置

# requirements.txt torch>=1.9.0 torchvision>=0.10.0 opencv-python>=4.5.0 numpy>=1.21.0 PyQt5>=5.15.0 mysql-connector-python>=8.0.0 ultralytics>=8.0.0 # YOLOv8 pillow>=8.3.0 scipy>=1.7.0 matplotlib>=3.4.0 seaborn>=0.11.0 pandas>=1.3.0

2.3 数据库配置

-- 创建数据库和表结构 CREATE DATABASE sheep_behavior; USE sheep_behavior; CREATE TABLE detection_records ( id INT AUTO_INCREMENT PRIMARY KEY, timestamp DATETIME, camera_id VARCHAR(50), sheep_id INT, behavior_type ENUM('eating', 'drinking', 'resting', 'walking', 'abnormal'), confidence FLOAT, image_path VARCHAR(255), bbox_x INT, bbox_y INT, bbox_width INT, bbox_height INT ); CREATE TABLE behavior_statistics ( id INT AUTO_INCREMENT PRIMARY KEY, date DATE, sheep_id INT, eating_duration INT, drinking_duration INT, resting_duration INT, walking_duration INT, abnormal_count INT );

3. YOLO模型选择与训练

3.1 数据集准备与标注

羊只行为检测数据集应包含以下类别：

• eating（进食）
• drinking（饮水）
• resting（休息）
• walking（行走）
• abnormal（异常行为）

# 数据集目录结构 dataset/ ├── images/ │ ├── train/ │ ├── val/ │ └── test/ └── labels/ ├── train/ ├── val/ └── test/ # YOLO格式标注示例 # class x_center y_center width height 0 0.5 0.5 0.3 0.4 1 0.7 0.3 0.2 0.3

3.2 YOLOv5模型训练配置

# data/sheep_behavior.yaml path: /path/to/dataset train: images/train val: images/val test: images/test nc: 5 # number of classes names: ['eating', 'drinking', 'resting', 'walking', 'abnormal'] # 模型配置文件修改 # models/yolov5s.yaml nc: 5 # 修改类别数

3.3 训练脚本实现

# train.py import torch from ultralytics import YOLO import argparse def train_yolov5_model(): # 加载预训练模型 model = YOLO('yolov5s.pt') # 训练参数配置 results = model.train( data='data/sheep_behavior.yaml', epochs=100, imgsz=640, batch_size=16, device='0', # 使用GPU workers=4, patience=10, save=True, exist_ok=True ) return results def train_yolov8_model(): model = YOLO('yolov8n.pt') results = model.train( data='data/sheep_behavior.yaml', epochs=100, imgsz=640, batch_size=16, 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 ) return results if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument('--model', type=str, default='yolov5', choices=['yolov5', 'yolov8', 'yolov11']) args = parser.parse_args() if args.model == 'yolov5': train_yolov5_model() elif args.model == 'yolov8': train_yolov8_model()

4. PyQt图形界面设计

4.1 主界面布局设计

# main_window.py import sys from PyQt5.QtWidgets import (QApplication, QMainWindow, QVBoxLayout, QHBoxLayout, QWidget, QLabel, QPushButton, QTabWidget, QTextEdit, QGroupBox, QComboBox, QSpinBox, QDoubleSpinBox, QCheckBox) from PyQt5.QtCore import Qt, QTimer, pyqtSignal from PyQt5.QtGui import QImage, QPixmap import cv2 import mysql.connector class MainWindow(QMainWindow): def __init__(self): super().__init__() self.init_ui() self.setup_database() self.setup_camera() def init_ui(self): self.setWindowTitle("牧场羊行为检测系统") 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: 1px solid gray;") self.video_label.setAlignment(Qt.AlignCenter) left_layout.addWidget(self.video_label) # 控制按钮 control_layout = QHBoxLayout() self.start_btn = QPushButton("开始检测") self.stop_btn = QPushButton("停止检测") self.settings_btn = QPushButton("设置") control_layout.addWidget(self.start_btn) control_layout.addWidget(self.stop_btn) control_layout.addWidget(self.settings_btn) left_layout.addLayout(control_layout) # 右侧信息显示区域 right_layout = QVBoxLayout() # 实时检测结果 result_group = QGroupBox("实时检测结果") result_layout = QVBoxLayout() self.result_text = QTextEdit() self.result_text.setMaximumHeight(200) result_layout.addWidget(self.result_text) result_group.setLayout(result_layout) right_layout.addWidget(result_group) # 统计信息 stats_group = QGroupBox("行为统计") stats_layout = QVBoxLayout() self.stats_text = QTextEdit() stats_layout.addWidget(self.stats_text) stats_group.setLayout(stats_layout) right_layout.addWidget(stats_group) # 系统状态 status_group = QGroupBox("系统状态") status_layout = QVBoxLayout() self.status_text = QTextEdit() self.status_text.setMaximumHeight(100) status_layout.addWidget(self.status_text) status_group.setLayout(status_layout) right_layout.addWidget(status_group) main_layout.addLayout(left_layout, 2) main_layout.addLayout(right_layout, 1) # 连接信号槽 self.start_btn.clicked.connect(self.start_detection) self.stop_btn.clicked.connect(self.stop_detection) # 定时器更新视频 self.timer = QTimer() self.timer.timeout.connect(self.update_frame) def setup_database(self): try: self.db_connection = mysql.connector.connect( host='localhost', user='root', password='password', database='sheep_behavior' ) self.status_text.append("数据库连接成功") except Exception as e: self.status_text.append(f"数据库连接失败: {str(e)}") def setup_camera(self): self.cap = cv2.VideoCapture(0) # 默认摄像头 if not self.cap.isOpened(): self.status_text.append("摄像头初始化失败") def start_detection(self): self.timer.start(30) # 30ms更新一帧 self.status_text.append("开始检测...") def stop_detection(self): self.timer.stop() self.status_text.append("停止检测") def update_frame(self): ret, frame = self.cap.read() if ret: # 转换为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) self.video_label.setPixmap(QPixmap.fromImage(qt_image)) # 进行行为检测 self.detect_behavior(frame) def detect_behavior(self, frame): # 这里调用YOLO模型进行检测 # 伪代码示例 results = self.model(frame) # YOLO检测 self.process_detection_results(results) def process_detection_results(self, results): # 处理检测结果并更新界面 detection_info = "" for result in results: detection_info += f"羊只ID: {result.sheep_id}, " detection_info += f"行为: {result.behavior}, " detection_info += f"置信度: {result.confidence:.2f}\n" self.result_text.setText(detection_info) self.save_to_database(results) if __name__ == "__main__": app = QApplication(sys.argv) window = MainWindow() window.show() sys.exit(app.exec_())

5. 数据库集成与数据管理

5.1 数据库操作类

# database_manager.py import mysql.connector from datetime import datetime import json class DatabaseManager: def __init__(self, host='localhost', user='root', password='password', database='sheep_behavior'): self.connection = mysql.connector.connect( host=host, user=user, password=password, database=database ) self.cursor = self.connection.cursor() def insert_detection_record(self, camera_id, sheep_id, behavior_type, confidence, image_path, bbox): """插入检测记录""" query = """ INSERT INTO detection_records (timestamp, camera_id, sheep_id, behavior_type, confidence, image_path, bbox_x, bbox_y, bbox_width, bbox_height) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """ values = (datetime.now(), camera_id, sheep_id, behavior_type, confidence, image_path, bbox[0], bbox[1], bbox[2], bbox[3]) self.cursor.execute(query, values) self.connection.commit() def get_daily_statistics(self, date): """获取每日统计信息""" query = """ SELECT behavior_type, COUNT(*) as count, AVG(confidence) as avg_confidence FROM detection_records WHERE DATE(timestamp) = %s GROUP BY behavior_type """ self.cursor.execute(query, (date,)) return self.cursor.fetchall() def get_sheep_behavior_history(self, sheep_id, days=7): """获取指定羊只的行为历史""" query = """ SELECT DATE(timestamp) as date, behavior_type, COUNT(*) as count FROM detection_records WHERE sheep_id = %s AND timestamp >= DATE_SUB(NOW(), INTERVAL %s DAY) GROUP BY DATE(timestamp), behavior_type ORDER BY date DESC """ self.cursor.execute(query, (sheep_id, days)) return self.cursor.fetchall() def close(self): """关闭数据库连接""" self.cursor.close() self.connection.close()

5.2 数据备份与恢复

# backup_manager.py import shutil import json from datetime import datetime import os class BackupManager: def __init__(self, db_manager, backup_dir='backups'): self.db_manager = db_manager self.backup_dir = backup_dir os.makedirs(backup_dir, exist_ok=True) def create_backup(self): """创建数据库备份""" timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") backup_file = os.path.join(self.backup_dir, f"backup_{timestamp}.sql") # 使用mysqldump创建备份 import subprocess cmd = [ 'mysqldump', '-u', self.db_manager.user, f'-p{self.db_manager.password}', self.db_manager.database ] with open(backup_file, 'w') as f: subprocess.run(cmd, stdout=f) return backup_file def restore_backup(self, backup_file): """恢复数据库备份""" import subprocess cmd = [ 'mysql', '-u', self.db_manager.user, f'-p{self.db_manager.password}', self.db_manager.database ] with open(backup_file, 'r') as f: subprocess.run(cmd, stdin=f)

6. 模型优化与性能提升

6.1 多模型集成策略

# model_ensemble.py import torch import numpy as np from collections import defaultdict class ModelEnsemble: def __init__(self, model_paths): self.models = [] for path in model_paths: if path.endswith('.pt'): model = torch.hub.load('ultralytics/yolov5', 'custom', path=path) else: from ultralytics import YOLO model = YOLO(path) self.models.append(model) def predict(self, image, confidence_threshold=0.5, iou_threshold=0.5): """多模型集成预测""" all_predictions = [] for model in self.models: if hasattr(model, 'predict'): # YOLOv8 results = model.predict(image, conf=confidence_threshold, iou=iou_threshold) predictions = results[0].boxes if len(results) > 0 else [] else: # YOLOv5 results = model(image) predictions = results.xyxy[0].cpu().numpy() all_predictions.extend(predictions) # 使用加权投票法集成结果 return self.weighted_voting(all_predictions) def weighted_voting(self, predictions): """加权投票集成""" if not predictions: return [] # 根据置信度进行加权 weighted_boxes = defaultdict(list) for pred in predictions: if len(pred) >= 6: # YOLOv5格式: [x1, y1, x2, y2, conf, class] x1, y1, x2, y2, conf, cls = pred[:6] center_x = (x1 + x2) / 2 center_y = (y1 + y2) / 2 key = (int(cls), int(center_x//10), int(center_y//10)) weighted_boxes[key].append((pred, conf)) # 合并重叠的检测框 final_predictions = [] for key, boxes in weighted_boxes.items(): if boxes: # 按置信度加权平均 total_conf = sum(conf for _, conf in boxes) weighted_box = np.zeros(6) for box, conf in boxes: weight = conf / total_conf weighted_box += box * weight final_predictions.append(weighted_box) return final_predictions

6.2 模型量化与加速

# model_optimizer.py import torch import torch.nn as nn from torch.quantization import quantize_dynamic class ModelOptimizer: def __init__(self, model_path): self.model = torch.load(model_path) def quantize_model(self): """动态量化模型""" # 量化模型 quantized_model = quantize_dynamic( self.model, {nn.Linear, nn.Conv2d}, dtype=torch.qint8 ) return quantized_model def optimize_for_inference(self): """优化模型推理速度""" self.model.eval() # 使用TorchScript优化 scripted_model = torch.jit.script(self.model) # 应用优化通道 with torch.no_grad(): optimized_model = torch.jit.optimize_for_inference(scripted_model) return optimized_model def save_optimized_model(self, output_path): """保存优化后的模型""" optimized_model = self.optimize_for_inference() torch.jit.save(optimized_model, output_path)

7. 系统部署与监控

7.1 系统服务化部署

# system_service.py import time import logging from threading import Thread from flask import Flask, jsonify, request class SystemService: def __init__(self, detection_system): self.detection_system = detection_system self.app = Flask(__name__) self.setup_routes() self.setup_logging() def setup_logging(self): logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('system.log'), logging.StreamHandler() ] ) self.logger = logging.getLogger(__name__) def setup_routes(self): @self.app.route('/api/start', methods=['POST']) def start_detection(): self.detection_system.start() return jsonify({'status': 'started'}) @self.app.route('/api/stop', methods=['POST']) def stop_detection(): self.detection_system.stop() return jsonify({'status': 'stopped'}) @self.app.route('/api/status', methods=['GET']) def get_status(): status = self.detection_system.get_status() return jsonify(status) @self.app.route('/api/statistics', methods=['GET']) def get_statistics(): days = request.args.get('days', 7, type=int) stats = self.detection_system.get_statistics(days) return jsonify(stats) def run(self, host='0.0.0.0', port=5000): self.app.run(host=host, port=port, debug=False) class HealthMonitor(Thread): def __init__(self, detection_system): super().__init__() self.detection_system = detection_system self.running = True def run(self): while self.running: # 监控系统健康状态 cpu_usage = self.get_cpu_usage() memory_usage = self.get_memory_usage() gpu_usage = self.get_gpu_usage() # 记录系统状态 self.log_system_status(cpu_usage, memory_usage, gpu_usage) # 检查系统异常 self.check_system_health(cpu_usage, memory_usage, gpu_usage) time.sleep(60) # 每分钟检查一次 def get_cpu_usage(self): import psutil return psutil.cpu_percent(interval=1) def get_memory_usage(self): import psutil return psutil.virtual_memory().percent def get_gpu_usage(self): try: import GPUtil gpus = GPUtil.getGPUs() return [gpu.load * 100 for gpu in gpus] except ImportError: return [] def log_system_status(self, cpu, memory, gpu): logging.info(f"CPU: {cpu}%, Memory: {memory}%, GPU: {gpu}") def check_system_health(self, cpu, memory, gpu): if cpu > 90 or memory > 90: logging.warning("系统资源使用率过高") if any(usage > 90 for usage in gpu): logging.warning("GPU使用率过高")

7.2 性能监控与告警

# performance_monitor.py import time import smtplib from email.mime.text import MimeText from datetime import datetime class PerformanceMonitor: def __init__(self, config): self.config = config self.performance_data = [] self.alert_thresholds = { 'cpu_usage': 85, 'memory_usage': 85, 'gpu_usage': 90, 'detection_delay': 1000 # ms } def monitor_performance(self): """监控系统性能""" while True: metrics = self.collect_metrics() self.performance_data.append({ 'timestamp': datetime.now(), 'metrics': metrics }) # 检查是否需要发送告警 self.check_alerts(metrics) # 保留最近24小时数据 self.cleanup_old_data() time.sleep(300) # 每5分钟收集一次 def collect_metrics(self): """收集性能指标""" import psutil metrics = {} # CPU使用率 metrics['cpu_usage'] = psutil.cpu_percent(interval=1) # 内存使用率 metrics['memory_usage'] = psutil.virtual_memory().percent # 磁盘使用率 metrics['disk_usage'] = psutil.disk_usage('/').percent # 网络IO net_io = psutil.net_io_counters() metrics['network_sent'] = net_io.bytes_sent metrics['network_recv'] = net_io.bytes_recv return metrics def check_alerts(self, metrics): """检查性能告警""" alerts = [] if metrics['cpu_usage'] > self.alert_thresholds['cpu_usage']: alerts.append(f"CPU使用率过高: {metrics['cpu_usage']}%") if metrics['memory_usage'] > self.alert_thresholds['memory_usage']: alerts.append(f"内存使用率过高: {metrics['memory_usage']}%") if alerts: self.send_alert(alerts) def send_alert(self, alerts): """发送告警邮件""" try: msg = MimeText('\n'.join(alerts)) msg['Subject'] = '牧场检测系统告警' msg['From'] = self.config['email_from'] msg['To'] = self.config['email_to'] server = smtplib.SMTP(self.config['smtp_server'], self.config['smtp_port']) server.starttls() server.login(self.config['email_user'], self.config['email_password']) server.send_message(msg) server.quit() except Exception as e: logging.error(f"发送告警邮件失败: {e}")

8. 实际应用案例与效果分析

8.1 部署实施步骤

1. 硬件部署

   • 摄像头安装位置选择（覆盖饮水区、进食区、休息区）
   • 网络布线及电源供应
   • 服务器机房环境准备

2. 软件部署

   • 操作系统及驱动安装
   • Python环境配置
   • 数据库初始化
   • 模型文件部署

3. 系统调试

   • 摄像头角度调整
   • 光照条件优化
   • 模型参数调优
   • 系统稳定性测试

8.2 效果评估指标

# evaluation_metrics.py import numpy as np from sklearn.metrics import precision_score, recall_score, f1_score, confusion_matrix class EvaluationMetrics: def __init__(self, true_labels, predicted_labels, class_names): self.true_labels = true_labels self.predicted_labels = predicted_labels self.class_names = class_names def calculate_metrics(self): """计算各项评估指标""" metrics = {} # 精确率、召回率、F1分数 metrics['precision'] = precision_score( self.true_labels, self.predicted_labels, average='weighted' ) metrics['recall'] = recall_score( self.true_labels, self.predicted_labels, average='weighted' ) metrics['f1_score'] = f1_score( self.true_labels, self.predicted_labels, average='weighted' ) # 混淆矩阵 metrics['confusion_matrix'] = confusion_matrix( self.true_labels, self.predicted_labels ) # 各类别准确率 class_accuracy = {} cm = metrics['confusion_matrix'] for i, class_name in enumerate(self.class_names): if cm[i].sum() > 0: class_accuracy[class_name] = cm[i,i] / cm[i].sum() else: class_accuracy[class_name] = 0 metrics['class_accuracy'] = class_accuracy return metrics def generate_report(self): """生成评估报告""" metrics = self.calculate_metrics() report = "羊行为检测系统评估报告\n" report += "=" * 50 + "\n" report += f"总体精确率: {metrics['precision']:.3f}\n" report += f"总体召回率: {metrics['recall']:.3f}\n" report += f"F1分数: {metrics['f1_score']:.3f}\n\n" report += "各类别准确率:\n" for class_name, accuracy in metrics['class_accuracy'].items(): report += f"{class_name}: {accuracy:.3f}\n" return report

8.3 持续优化建议

1. 模型优化方向

   • 增加训练数据多样性
   • 尝试不同的数据增强策略
   • 优化模型超参数
   • 集成更多先进的目标检测算法

2. 系统优化方向

   • 实现分布式部署
   • 优化数据库查询性能
   • 增加缓存机制
   • 实现负载均衡

3. 功能扩展方向

   • 增加羊只个体识别功能
   • 实现行为预测分析
   • 集成环境传感器数据
   • 开发移动端应用

本系统通过结合先进的YOLO目标检测算法、PyQt图形界面和MySQL数据库，为牧场管理者提供了一个完整的羊只行为监测解决方案。系统具有良好的可扩展性和稳定性，能够有效提升牧场管理的智能化水平。