在牧场智能化管理系统中羊只行为检测是一个关键技术需求。传统的人工观察方式效率低下且容易遗漏重要行为信息。本文将详细介绍基于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 SSD2.2 Python环境配置# requirements.txt torch1.9.0 torchvision0.10.0 opencv-python4.5.0 numpy1.21.0 PyQt55.15.0 mysql-connector-python8.0.0 ultralytics8.0.0 # YOLOv8 pillow8.3.0 scipy1.7.0 matplotlib3.4.0 seaborn0.11.0 pandas1.3.02.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.33.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( datadata/sheep_behavior.yaml, epochs100, imgsz640, batch_size16, device0, # 使用GPU workers4, patience10, saveTrue, exist_okTrue ) return results def train_yolov8_model(): model YOLO(yolov8n.pt) results model.train( datadata/sheep_behavior.yaml, epochs100, imgsz640, batch_size16, lr00.01, lrf0.01, momentum0.937, weight_decay0.0005, warmup_epochs3.0, warmup_momentum0.8, box7.5, cls0.5, dfl1.5 ) return results if __name__ __main__: parser argparse.ArgumentParser() parser.add_argument(--model, typestr, defaultyolov5, 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( hostlocalhost, userroot, passwordpassword, databasesheep_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, hostlocalhost, userroot, passwordpassword, databasesheep_behavior): self.connection mysql.connector.connect( hosthost, useruser, passwordpassword, databasedatabase ) 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, days7): 获取指定羊只的行为历史 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_dirbackups): self.db_manager db_manager self.backup_dir backup_dir os.makedirs(backup_dir, exist_okTrue) def create_backup(self): 创建数据库备份 timestamp datetime.now().strftime(%Y%m%d_%H%M%S) backup_file os.path.join(self.backup_dir, fbackup_{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, stdoutf) 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, stdinf)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, pathpath) else: from ultralytics import YOLO model YOLO(path) self.models.append(model) def predict(self, image, confidence_threshold0.5, iou_threshold0.5): 多模型集成预测 all_predictions [] for model in self.models: if hasattr(model, predict): # YOLOv8 results model.predict(image, confconfidence_threshold, iouiou_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_predictions6.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}, dtypetorch.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( levellogging.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, typeint) stats self.detection_system.get_statistics(days) return jsonify(stats) def run(self, host0.0.0.0, port5000): self.app.run(hosthost, portport, debugFalse) 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(interval1) 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(fCPU: {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(interval1) # 内存使用率 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(fCPU使用率过高: {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 部署实施步骤硬件部署摄像头安装位置选择覆盖饮水区、进食区、休息区网络布线及电源供应服务器机房环境准备软件部署操作系统及驱动安装Python环境配置数据库初始化模型文件部署系统调试摄像头角度调整光照条件优化模型参数调优系统稳定性测试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, averageweighted ) metrics[recall] recall_score( self.true_labels, self.predicted_labels, averageweighted ) metrics[f1_score] f1_score( self.true_labels, self.predicted_labels, averageweighted ) # 混淆矩阵 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 fF1分数: {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 report8.3 持续优化建议模型优化方向增加训练数据多样性尝试不同的数据增强策略优化模型超参数集成更多先进的目标检测算法系统优化方向实现分布式部署优化数据库查询性能增加缓存机制实现负载均衡功能扩展方向增加羊只个体识别功能实现行为预测分析集成环境传感器数据开发移动端应用本系统通过结合先进的YOLO目标检测算法、PyQt图形界面和MySQL数据库为牧场管理者提供了一个完整的羊只行为监测解决方案。系统具有良好的可扩展性和稳定性能够有效提升牧场管理的智能化水平。