From 4efc605092f68311876299290a33697ebbfd3a5d Mon Sep 17 00:00:00 2001
From: liurunyu <lry9898@163.com>
Date: 星期三, 25 六月 2025 15:28:16 +0800
Subject: [PATCH] 水肥、气象、墒情数据处理：最新数据、历史数据、墒情日数据
---
pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/pSdV1/ProtocolParserSdV1.java | 202 ++++++++++++++++++++++++++++++++------------------
1 files changed, 129 insertions(+), 73 deletions(-)
diff --git a/pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/pSdV1/ProtocolParserSdV1.java b/pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/pSdV1/ProtocolParserSdV1.java
index 1ca98b2..7ba7e80 100644
--- a/pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/pSdV1/ProtocolParserSdV1.java
+++ b/pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/pSdV1/ProtocolParserSdV1.java
@@ -3,9 +3,16 @@
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.dy.common.mw.protocol.Command;
-import com.dy.common.mw.protocol4Mqtt.pSdV1.downVos.ComCtrlVo;
+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
@@ -13,44 +20,106 @@
* @Description
*/
public class ProtocolParserSdV1 {
- public MqttSubMsgSdV1 parseSubMsg(String deviceId, String topic, MqttMessage mqttMsg) throws Exception {
- MqttSubMsgSdV1 ms = new MqttSubMsgSdV1(deviceId, topic, new String(mqttMsg.getPayload(), "UTF-8"));
- return ms;
+ 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("鎺ユ敹鍒癕QTT娑堟伅锛屽崗璁�" + subTopic.protocol + "锛岃澶嘔D" + subTopic.devId + "锛屼富棰�" + subTopic.longName() + "娑堟伅鏍煎紡闈瀓son鏁版嵁(" + 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("鎺ユ敹鍒癕QTT娑堟伅锛屽崗璁�" + subTopic.protocol + "锛岃澶嘔D" + subTopic.devId + "锛屼富棰�" + subTopic.type + "娑堟伅瑙ｆ瀽閫昏緫鏈疄鐜�");
+ }
+ }
+ vo.setNo(subTopic.no) ;
+ msg.vo4Up = vo ;
+ callback.callback(msg);
+ callback.notify(msg.deviceId, stInfo);
+ return msg;
}
- public MqttPubMsgSdV1 createPubMsg(String orgTag, Command com) throws Exception {
- MqttPubMsgSdV1 msg = null ;
- switch (com.code){
- case CodeSdV1.cd_Fault:{
+ 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 ;
+ msg = this.createPubMsgOfFault(orgTag, com);
+ break;
}
- case CodeSdV1.cd_Stir:{
+ case CodeSdV1.cd_Stir -> {
//鎼呮媽鍚仠鍛戒护
this.checkParam(com);
this.checkRtnWebUrl(com);
- msg = this.createPubMsgOfStir(orgTag, com) ;
- break ;
+ msg = this.createPubMsgOfStir(orgTag, com);
+ break;
}
- case CodeSdV1.cd_Inject:{
+ case CodeSdV1.cd_Inject -> {
//娉ㄨ偉鍚仠鍛戒护
this.checkParam(com);
this.checkRtnWebUrl(com);
- msg = this.createPubMsgOfInject(orgTag, com) ;
- break ;
+ msg = this.createPubMsgOfInject(orgTag, com);
+ break;
}
- case CodeSdV1.cd_Irr:{
+ case CodeSdV1.cd_Irr -> {
//鐏屾簤鍚仠鍛戒护
this.checkParam(com);
this.checkRtnWebUrl(com);
- msg = this.createPubMsgOfIrr(orgTag, com) ;
- break ;
+ msg = this.createPubMsgOfIrr(orgTag, com);
+ break;
}
- default:{
- throw new Exception("鎺ユ敹鍒癕QTT鍛戒护锛屽崗璁�" + com.protocol + "鐗堟湰" + com.protocolVersion + "鍔熻兘鐮�" + com.code + "鏋勯�犲櫒鏈疄鐜�") ;
+ case CodeSdV1.cd_Param -> {
+ //璁剧疆鍙傛暟
+ this.checkParam(com);
+ this.checkRtnWebUrl(com);
+ msg = this.createPubMsgOfParam(orgTag, com);
+ break;
+ }
+ default -> {
+ throw new Exception("鎺ユ敹鍒癕QTT鍛戒护锛屽崗璁�" + com.protocol + "鐗堟湰" + com.protocolVersion + "鍔熻兘鐮�" + com.code + "鏋勯�犲櫒鏈疄鐜�");
}
}
return msg ;
@@ -65,106 +134,93 @@
throw new Exception("鎺ユ敹鍒癕QTT鍛戒护锛屽崗璁�" + com.protocol + "鐗堟湰" + com.protocolVersion + "鍔熻兘鐮�" + com.code + "鍛戒护缁撴灉鍥炴敹URL涓虹┖") ;
}
}
- private MqttPubMsgSdV1 createPubMsgOfFault(String orgTag, Command com) throws Exception {
+ 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杞珻omCtrlVo涓簄ull") ;
}
- MqttPubMsgSdV1 msg = new MqttPubMsgSdV1() ;
+ MqttPubMsg msg = new MqttPubMsg() ;
this.setPubMsgBase(com, msg);
msg.isCacheForOffLine = false ;
msg.hasResponse = true ;
- msg.address = 123 ;
- msg.value = "" + (cvo.isDo?1:0);
- msg.topic = createTopic(orgTag, com) ;
- msg.msg = "" ;
+ 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 MqttPubMsgSdV1 createPubMsgOfStir(String orgTag, Command com) throws Exception {
+ 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杞珻omCtrlVo涓簄ull") ;
}
- MqttPubMsgSdV1 msg = new MqttPubMsgSdV1() ;
+ MqttPubMsg msg = new MqttPubMsg() ;
this.setPubMsgBase(com, msg);
msg.isCacheForOffLine = false ;
msg.hasResponse = true ;
- msg.address = 123 ;
- msg.value = "" + (cvo.isDo?1:0);
- msg.topic = createTopic(orgTag, com) ;
- msg.msg = "" ;
+ 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 MqttPubMsgSdV1 createPubMsgOfInject(String orgTag, Command com) throws Exception {
+ 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杞珻omCtrlVo涓簄ull") ;
}
- MqttPubMsgSdV1 msg = new MqttPubMsgSdV1() ;
+ MqttPubMsg msg = new MqttPubMsg() ;
this.setPubMsgBase(com, msg);
msg.isCacheForOffLine = false ;
msg.hasResponse = true ;
- msg.address = 123 ;
- msg.value = "" + (cvo.isDo?1:0);
- msg.topic = createTopic(orgTag, com) ;
- msg.msg = "" ;
+ 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 MqttPubMsgSdV1 createPubMsgOfIrr(String orgTag, Command com) throws Exception {
+ 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杞珻omCtrlVo涓簄ull") ;
}
- MqttPubMsgSdV1 msg = new MqttPubMsgSdV1() ;
+ MqttPubMsg msg = new MqttPubMsg() ;
this.setPubMsgBase(com, msg);
msg.isCacheForOffLine = false ;
msg.hasResponse = true ;
- msg.address = 123 ;
- msg.value = "" + (cvo.isDo?1:0);
- msg.topic = createTopic(orgTag, com) ;
- msg.msg = "" ;
+ 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 void setPubMsgBase(Command com, MqttPubMsgSdV1 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杞珻omSetParamVo涓簄ull") ;
+ }
+ 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 ;
}
- private String createTopic(String orgTag, Command com){
- String topic = null ;
- switch (com.code){
- case CodeSdV1.cd_Fault:{
- topic = orgTag + "/" + com.protocol + com.protocolVersion + "/" + com.rtuAddr + "/control/m4" ;
- break ;
- }
- case CodeSdV1.cd_Stir:{
- topic = orgTag + "/" + com.protocol + com.protocolVersion + "/" + com.rtuAddr + "/control/m80" ;
- break ;
- }
- case CodeSdV1.cd_Inject:{
- topic = orgTag + "/" + com.protocol + com.protocolVersion + "/" + com.rtuAddr + "/control/m1" ;
- break ;
- }
- case CodeSdV1.cd_Irr:{
- topic = orgTag + "/" + com.protocol + com.protocolVersion + "/" + com.rtuAddr + "/control/m2" ;
- break ;
- }
- default:{
- topic = null ;
- break;
- }
- }
- return topic ;
- }
}
--
Gitblit v1.8.0