package com.dy.common.mw.protocol4Mqtt.pSdV1;
|
|
import com.alibaba.fastjson2.JSON;
|
import com.alibaba.fastjson2.JSONObject;
|
import com.dy.common.mw.protocol.Command;
|
import com.dy.common.mw.protocol4Mqtt.*;
|
import com.dy.common.mw.protocol4Mqtt.pSdV1.comParam.ComCtrlVo;
|
import com.dy.common.mw.protocol4Mqtt.pSdV1.comParam.ComSetParamVo;
|
import com.dy.common.mw.protocol4Mqtt.pSdV1.downVos.FaultClearVo;
|
import com.dy.common.mw.protocol4Mqtt.pSdV1.downVos.InjectStartVo;
|
import com.dy.common.mw.protocol4Mqtt.pSdV1.downVos.ParamSetVo;
|
import com.dy.common.mw.protocol4Mqtt.pSdV1.downVos.StirStartVo;
|
import com.dy.common.mw.protocol4Mqtt.pSdV1.upVos.*;
|
import com.dy.common.mw.protocol4Mqtt.status.DevRunInfo;
|
import org.eclipse.paho.client.mqttv3.MqttMessage;
|
|
/**
|
* @Author: liurunyu
|
* @Date: 2025/6/5 11:41
|
* @Description
|
*/
|
public class ProtocolParserSdV1 {
|
public MqttSubMsg parseSubMsg(MqttTopic subTopic, MqttMessage mqttMsg, MqttCallback callback) throws Exception {
|
String strTxt = new String(mqttMsg.getPayload(), "UTF-8");
|
if(!JSON.isValid(strTxt)){
|
throw new Exception("接收到MQTT消息，协议" + subTopic.protocol + "，设备ID" + subTopic.devId + "，主题" + subTopic.longName() + "消息格式非json数据(" + strTxt + ")") ;
|
}
|
MqttSubMsg msg = new MqttSubMsg(subTopic, strTxt);
|
Vo4Up vo ;
|
DevRunInfo stInfo = null ;
|
switch (subTopic.type) {
|
case ProtocolConstantSdV1.SubTopicWeather -> {
|
WeatherVo wvo = JSON.parseObject(strTxt, WeatherVo.class);
|
//调整小数
|
wvo.airTemperature = (wvo.airTemperature == null ? null: (wvo.airTemperature / 10.0F)) ;
|
wvo.airHumidity = (wvo.airHumidity == null ? null: (wvo.airHumidity / 10.0F)) ;
|
wvo.rainfall = (wvo.rainfall == null ? null: (wvo.rainfall / 10.0F)) ;
|
wvo.windSpeed = (wvo.windSpeed == null ? null: (wvo.windSpeed / 10.0F)) ;
|
vo = wvo ;
|
break;
|
}
|
case ProtocolConstantSdV1.SubTopicSoil -> {
|
SoilVo svo = JSON.parseObject(strTxt, SoilVo.class);
|
//调整小数
|
svo.soilHumidity1 = (svo.soilHumidity1 == null ? null: (svo.soilHumidity1 / 10.0F)) ;
|
svo.soilHumidity2 = (svo.soilHumidity2 == null ? null: (svo.soilHumidity2 / 10.0F)) ;
|
svo.soilHumidity3 = (svo.soilHumidity3 == null ? null: (svo.soilHumidity3 / 10.0F)) ;
|
svo.soilHumidity4 = (svo.soilHumidity4 == null ? null: (svo.soilHumidity4 / 10.0F)) ;
|
svo.soilHumidity5 = (svo.soilHumidity5 == null ? null: (svo.soilHumidity5 / 10.0F)) ;
|
svo.soilTemperature1 = (svo.soilTemperature1 == null ? null: (svo.soilTemperature1 / 10.0F)) ;
|
svo.soilTemperature2 = (svo.soilTemperature2 == null ? null: (svo.soilTemperature2 / 10.0F)) ;
|
svo.soilTemperature3 = (svo.soilTemperature3 == null ? null: (svo.soilTemperature3 / 10.0F)) ;
|
svo.soilTemperature4 = (svo.soilTemperature4 == null ? null: (svo.soilTemperature4 / 10.0F)) ;
|
svo.soilTemperature5 = (svo.soilTemperature5 == null ? null: (svo.soilTemperature5 / 10.0F)) ;
|
vo = svo ;
|
break;
|
}
|
case ProtocolConstantSdV1.SubTopicManure -> {
|
ManureVo mvo = JSON.parseObject(strTxt, ManureVo.class);
|
//调整小数
|
mvo.manureFlow = (mvo.manureFlow == null ? null: (mvo.manureFlow / 10.0F)) ;
|
vo = mvo ;
|
|
stInfo = new DevRunInfo() ;
|
stInfo.devId = msg.deviceId ;
|
stInfo.stirRunning = (mvo.stirRunning1 ==null?false:(mvo.stirRunning1.byteValue()==1?true:false)) ; //搅拌运行 true是 false否
|
stInfo.injectRunning = (mvo.injectRunning==null?false:(mvo.injectRunning.byteValue()==1?true:false)) ; //注肥运行 true是 false否
|
stInfo.irrRunning = (mvo.irrRunning==null?false:(mvo.irrRunning.byteValue()==1?true:false)) ; //灌溉运行 true是 false否
|
stInfo.alarm = (mvo.alarm==null?false:(mvo.alarm.byteValue()==1?true:false)) ; //报警 true是 false否
|
break;
|
}
|
default -> {
|
throw new Exception("接收到MQTT消息，协议" + subTopic.protocol + "，设备ID" + subTopic.devId + "，主题" + subTopic.type + "消息解析逻辑未实现");
|
}
|
}
|
vo.setNo(subTopic.no) ;
|
msg.vo4Up = vo ;
|
callback.callback(msg);
|
callback.notify(msg.deviceId, stInfo);
|
return msg;
|
}
|
|
public MqttPubMsg createPubMsg(String orgTag, Command com) throws Exception {
|
MqttPubMsg msg ;
|
switch (com.code) {
|
case CodeSdV1.cd_Fault -> {
|
//故障解除命令
|
this.checkParam(com);
|
this.checkRtnWebUrl(com);
|
msg = this.createPubMsgOfFault(orgTag, com);
|
break;
|
}
|
case CodeSdV1.cd_Stir -> {
|
//搅拌启停命令
|
this.checkParam(com);
|
this.checkRtnWebUrl(com);
|
msg = this.createPubMsgOfStir(orgTag, com);
|
break;
|
}
|
case CodeSdV1.cd_Inject -> {
|
//注肥启停命令
|
this.checkParam(com);
|
this.checkRtnWebUrl(com);
|
msg = this.createPubMsgOfInject(orgTag, com);
|
break;
|
}
|
case CodeSdV1.cd_Irr -> {
|
//灌溉启停命令
|
this.checkParam(com);
|
this.checkRtnWebUrl(com);
|
msg = this.createPubMsgOfIrr(orgTag, com);
|
break;
|
}
|
case CodeSdV1.cd_Param -> {
|
//设置参数
|
this.checkParam(com);
|
this.checkRtnWebUrl(com);
|
msg = this.createPubMsgOfParam(orgTag, com);
|
break;
|
}
|
default -> {
|
throw new Exception("接收到MQTT命令，协议" + com.protocol + "版本" + com.protocolVersion + "功能码" + com.code + "构造器未实现");
|
}
|
}
|
return msg ;
|
}
|
private void checkParam(Command com)throws Exception {
|
if(com.param == null){
|
throw new Exception("接收到MQTT命令，协议" + com.protocol + "版本" + com.protocolVersion + "功能码" + com.code + "命令数据为空") ;
|
}
|
}
|
private void checkRtnWebUrl(Command com)throws Exception {
|
if(com.rtuResultSendWebUrl == null || com.rtuResultSendWebUrl.trim().equals("")){
|
throw new Exception("接收到MQTT命令，协议" + com.protocol + "版本" + com.protocolVersion + "功能码" + com.code + "命令结果回收URL为空") ;
|
}
|
}
|
private MqttPubMsg createPubMsgOfFault(String orgTag, Command com) throws Exception {
|
JSONObject obj = (JSONObject) com.param;
|
String json = obj.toJSONString();
|
ComCtrlVo cvo = JSON.parseObject(json, ComCtrlVo.class);
|
if(cvo == null){
|
throw new Exception("json转ComCtrlVo为null") ;
|
}
|
MqttPubMsg msg = new MqttPubMsg() ;
|
this.setPubMsgBase(com, msg);
|
msg.isCacheForOffLine = false ;
|
msg.hasResponse = true ;
|
msg.cd = CodeSdV1.cd_Fault ;
|
msg.topic = new MqttTopic(orgTag, com.protocol + com.protocolVersion, com.rtuAddr, ProtocolConstantSdV1.PubTopicFault, cvo.no) ;
|
msg.msg = JSON.toJSONString(new FaultClearVo(cvo.startTrueStopFalse ?(byte)1:0)) ;
|
return msg ;
|
}
|
private MqttPubMsg createPubMsgOfStir(String orgTag, Command com) throws Exception {
|
JSONObject obj = (JSONObject) com.param;
|
String json = obj.toJSONString();
|
ComCtrlVo cvo = JSON.parseObject(json, ComCtrlVo.class);
|
if(cvo == null){
|
throw new Exception("json转ComCtrlVo为null") ;
|
}
|
MqttPubMsg msg = new MqttPubMsg() ;
|
this.setPubMsgBase(com, msg);
|
msg.isCacheForOffLine = false ;
|
msg.hasResponse = true ;
|
msg.cd = CodeSdV1.cd_Fault ;
|
msg.topic = new MqttTopic(orgTag, com.protocol + com.protocolVersion, com.rtuAddr, ProtocolConstantSdV1.PubTopicStir, cvo.no) ;
|
msg.msg = JSON.toJSONString(new StirStartVo(cvo.startTrueStopFalse ?(byte)1:0)) ;
|
return msg ;
|
}
|
private MqttPubMsg createPubMsgOfInject(String orgTag, Command com) throws Exception {
|
JSONObject obj = (JSONObject) com.param;
|
String json = obj.toJSONString();
|
ComCtrlVo cvo = JSON.parseObject(json, ComCtrlVo.class);
|
if(cvo == null){
|
throw new Exception("json转ComCtrlVo为null") ;
|
}
|
MqttPubMsg msg = new MqttPubMsg() ;
|
this.setPubMsgBase(com, msg);
|
msg.isCacheForOffLine = false ;
|
msg.hasResponse = true ;
|
msg.cd = CodeSdV1.cd_Fault ;
|
msg.topic = new MqttTopic(orgTag, com.protocol + com.protocolVersion, com.rtuAddr, ProtocolConstantSdV1.PubTopicInject, cvo.no) ;
|
msg.msg = JSON.toJSONString(new InjectStartVo(cvo.startTrueStopFalse ?(byte)1:0)) ;
|
return msg ;
|
}
|
private MqttPubMsg createPubMsgOfIrr(String orgTag, Command com) throws Exception {
|
JSONObject obj = (JSONObject) com.param;
|
String json = obj.toJSONString();
|
ComCtrlVo cvo = JSON.parseObject(json, ComCtrlVo.class);
|
if(cvo == null){
|
throw new Exception("json转ComCtrlVo为null") ;
|
}
|
MqttPubMsg msg = new MqttPubMsg() ;
|
this.setPubMsgBase(com, msg);
|
msg.isCacheForOffLine = false ;
|
msg.hasResponse = true ;
|
msg.cd = CodeSdV1.cd_Fault ;
|
msg.topic = new MqttTopic(orgTag, com.protocol + com.protocolVersion, com.rtuAddr, ProtocolConstantSdV1.PubTopicIrr, cvo.no) ;
|
msg.msg = JSON.toJSONString(new StirStartVo(cvo.startTrueStopFalse ?(byte)1:0)) ;
|
return msg ;
|
}
|
|
private MqttPubMsg createPubMsgOfParam(String orgTag, Command com) throws Exception {
|
JSONObject obj = (JSONObject) com.param;
|
String json = obj.toJSONString();
|
ComSetParamVo cvo = JSON.parseObject(json, ComSetParamVo.class);
|
if(cvo == null){
|
throw new Exception("json转ComSetParamVo为null") ;
|
}
|
MqttPubMsg msg = new MqttPubMsg() ;
|
this.setPubMsgBase(com, msg);
|
msg.isCacheForOffLine = false ;
|
msg.hasResponse = true ;
|
msg.cd = CodeSdV1.cd_Param ;
|
msg.topic = new MqttTopic(orgTag, com.protocol + com.protocolVersion, com.rtuAddr, ProtocolConstantSdV1.PubTopicParam, cvo.no) ;
|
msg.msg = JSON.toJSONString(new ParamSetVo(cvo.stirDuration, cvo.injectDuration)) ;
|
return msg ;
|
}
|
|
private void setPubMsgBase(Command com, MqttPubMsg msg){
|
msg.commandId = com.id ;
|
msg.deviceId = com.rtuAddr ;
|
msg.mqttResultSendWebUrl = com.rtuResultSendWebUrl ;
|
}
|
|
|
}
|
