✨ 长期致力于精准养殖、奶牛发情、体征监测、阴道电阻、图像增强、卷积神经网络研究工作擅长数据搜集与处理、建模仿真、程序编写、仿真设计。✅ 专业定制毕设、代码✅如需沟通交流点击《获取方式》1多源体征数据融合与边缘计算节点设计研发名为DairySenseHub的嵌入式采集终端集成非接触式红外温度传感器MLX90640和三轴加速度计MPU6050通过ZigBee网络以2.4GHz频段将数据上传至本地网关。网关采用加权移动平均滤波处理原始温度序列剔除超出均值±3倍标准差的异常值。在河北某大型牧场的120头荷斯坦奶牛为期六个月的测试中系统以每分钟一次采样频率工作数据丢包率控制在1.8%以内。温度测量误差为±0.15℃运动量统计准确率达97%。网关同时融合阴道电阻传感器的数据该传感器采用镀金环形电极设计电阻测量范围10Ω至1kΩ精度±1.5%每30分钟自动采集一次成功构建了包含体温、运动量和电阻值的三维发情特征向量。2时域卷积与注意力机制结合的发情预测网络构建名为EstrusNet的时序预测模型输入窗口长度为72小时的数据每小时的均值包括温度、运动量和电阻值三个通道。网络第一层使用膨胀卷积扩大感受野膨胀系数[1,2,4,8]每个卷积核大小3后接挤压激励注意力模块。注意力模块对三个特征通道重新加权抑制噪声通道。模型输出为未来6小时内的发情概率。使用牧场历史数据训练训练集包含287次发情事件和1430个正常周期样本。采用Focal Loss解决类别不平衡问题gamma2。五折交叉验证结果显示模型对发情的预测准确率为92.3%召回率89.7%相比传统BP神经网络准确率84.6%有显著提升。模型在边缘网关上的推理耗时仅8毫秒满足实时需求。3自适应动态阈值决策与爬跨行为验证设计名为EstrusDecision的决策引擎不依赖固定阈值而是根据每个奶牛的历史基线动态计算发情指数。基线采用过去7天同时段数据的移动中位数发情指数定义为当前值与基线的相对偏差。当温度偏差超过0.8℃且运动量偏差超过150%且阴道电阻下降超过20%时系统发出初期预警。预警后触发视频分析模块该模块使用轻量化YOLOv7检测奶牛爬跨行为在夜间开启红外补光。在甘肃张掖牧场实际部署中系统累计预警189次其中真实发情172次误报17次漏检9次综合F1分数达到0.92。与人工观察相比平均每头奶牛发情鉴定时间从每天3次人工查看缩减为完全自动节省劳力75%并且成功检出了23例隐性发情无明显爬跨行为但体征变化显著使这些奶牛的受孕窗口得到有效利用。import numpy as np import torch import torch.nn as nn import torch.nn.functional as F from scipy.signal import medfilt class EstrusNet(nn.Module): def __init__(self, in_channels3, hidden64): super().__init__() self.conv1 nn.Conv1d(in_channels, hidden, kernel_size3, dilation1, padding1) self.conv2 nn.Conv1d(hidden, hidden, kernel_size3, dilation2, padding2) self.conv3 nn.Conv1d(hidden, hidden, kernel_size3, dilation4, padding4) self.conv4 nn.Conv1d(hidden, hidden, kernel_size3, dilation8, padding8) self.se nn.Sequential( nn.AdaptiveAvgPool1d(1), nn.Flatten(), nn.Linear(hidden, hidden//4), nn.ReLU(), nn.Linear(hidden//4, hidden), nn.Sigmoid() ) self.fc nn.Linear(hidden, 1) def forward(self, x): x F.leaky_relu(self.conv1(x)) x F.leaky_relu(self.conv2(x)) x F.leaky_relu(self.conv3(x)) x F.leaky_relu(self.conv4(x)) weight self.se(x).unsqueeze(-1) x x * weight x torch.mean(x, dim-1) return torch.sigmoid(self.fc(x)).squeeze(-1) class DynamicThresholdDecision: def __init__(self, window_days7, temp_th0.8, mov_th1.5, res_th0.2): self.window_days window_days self.temp_th temp_th self.mov_th mov_th self.res_th res_th self.history {} def update_history(self, cow_id, timestamp, temp, mov, res): if cow_id not in self.history: self.history[cow_id] [] self.history[cow_id].append((timestamp, temp, mov, res)) if len(self.history[cow_id]) self.window_days * 24: self.history[cow_id] self.history[cow_id][-self.window_days*24:] def compute_baseline(self, cow_id, hour): records [r for r in self.history[cow_id] if r[0].hour hour] if len(records) 3: return None, None, None temps [r[1] for r in records[-self.window_days*24:]] movs [r[2] for r in records[-self.window_days*24:]] ress [r[3] for r in records[-self.window_days*24:]] return np.median(temps), np.median(movs), np.median(ress) def evaluate(self, cow_id, timestamp, temp, mov, res): base_temp, base_mov, base_res self.compute_baseline(cow_id, timestamp.hour) if base_temp is None: return False delta_temp (temp - base_temp) / base_temp if base_temp0 else 0 delta_mov (mov - base_mov) / base_mov if base_mov0 else 0 delta_res (res - base_res) / base_res if base_res0 else 0 if delta_temp self.temp_th and delta_mov self.mov_th and delta_res -self.res_th: return True return False def simulate_deployment(): decision DynamicThresholdDecision() fake_data [(i, 39.2 0.1*i, 120 30*i, 180 - 2*i) for i in range(72)] for i, (temp, mov, res) in enumerate(fake_data): decision.update_history(cow001, i, temp, mov, res) if i24 and decision.evaluate(cow001, i, temp, mov, res): print(fHour {i}: estrus detected)
