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.MqttCallback; import com.dy.common.mw.protocol4Mqtt.MqttTopic; import com.dy.common.mw.protocol4Mqtt.Vo4Up; 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 MqttSubMsgSdV1 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 + ")") ; } MqttSubMsgSdV1 msg = new MqttSubMsgSdV1(subTopic, strTxt); Vo4Up vo ; DevRunInfo stInfo = null ; switch (subTopic.name) { case ProtocolConstantSdV1.SubTopicWeather -> { vo = JSON.parseObject(strTxt, WeatherVo.class); break; } case ProtocolConstantSdV1.SubTopicSoil -> { vo = JSON.parseObject(strTxt, SoilVo.class); break; } case ProtocolConstantSdV1.SubTopicManure -> { vo = JSON.parseObject(strTxt, ManureVo.class); break; } case ProtocolConstantSdV1.SubTopicState -> { //此处未完成,应该产生一些通信的info,供下面callback.notify(objs)通知出去 vo = JSON.parseObject(strTxt, StateVo.class); StateVo stVo = (StateVo)vo ; stInfo = new DevRunInfo() ; stInfo.devId = msg.deviceId ; stInfo.stirRunning = (stVo.stirRunning==null?false:(stVo.stirRunning.byteValue()==1?true:false)) ; //搅拌运行 true是 false否 stInfo.injectRunning = (stVo.injectRunning==null?false:(stVo.injectRunning.byteValue()==1?true:false)) ; //注肥运行 true是 false否 stInfo.irrRunning = (stVo.irrRunning==null?false:(stVo.irrRunning.byteValue()==1?true:false)) ; //灌溉运行 true是 false否 stInfo.alarm = (stVo.alarm==null?false:(stVo.alarm.byteValue()==1?true:false)) ; //报警 true是 false否 break; } default -> { throw new Exception("接收到MQTT消息,协议" + subTopic.protocol + ",设备ID" + subTopic.devId + ",主题" + subTopic.name + "消息解析逻辑未实现"); } } msg.vo4Up = vo ; callback.callback(msg); callback.notify(msg.deviceId, stInfo); return msg; } public MqttPubMsgSdV1 createPubMsg(String orgTag, Command com) throws Exception { MqttPubMsgSdV1 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 MqttPubMsgSdV1 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") ; } MqttPubMsgSdV1 msg = new MqttPubMsgSdV1() ; this.setPubMsgBase(com, msg); msg.isCacheForOffLine = false ; msg.hasResponse = true ; msg.cd = CodeSdV1.cd_Fault ; msg.topic = new MqttTopic(orgTag, com.protocol, com.rtuAddr, ProtocolConstantSdV1.PubTopicFault) ; msg.msg = JSON.toJSONString(new FaultClearVo(cvo.startTrueStopFalse ?(byte)1:0)) ; return msg ; } private MqttPubMsgSdV1 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") ; } MqttPubMsgSdV1 msg = new MqttPubMsgSdV1() ; this.setPubMsgBase(com, msg); msg.isCacheForOffLine = false ; msg.hasResponse = true ; msg.cd = CodeSdV1.cd_Fault ; msg.topic = new MqttTopic(orgTag, com.protocol, com.rtuAddr, ProtocolConstantSdV1.PubTopicStir) ; msg.msg = JSON.toJSONString(new StirStartVo(cvo.startTrueStopFalse ?(byte)1:0)) ; return msg ; } private MqttPubMsgSdV1 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") ; } MqttPubMsgSdV1 msg = new MqttPubMsgSdV1() ; this.setPubMsgBase(com, msg); msg.isCacheForOffLine = false ; msg.hasResponse = true ; msg.cd = CodeSdV1.cd_Fault ; msg.topic = new MqttTopic(orgTag, com.protocol, com.rtuAddr, ProtocolConstantSdV1.PubTopicInject) ; msg.msg = JSON.toJSONString(new InjectStartVo(cvo.startTrueStopFalse ?(byte)1:0)) ; return msg ; } private MqttPubMsgSdV1 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") ; } MqttPubMsgSdV1 msg = new MqttPubMsgSdV1() ; this.setPubMsgBase(com, msg); msg.isCacheForOffLine = false ; msg.hasResponse = true ; msg.cd = CodeSdV1.cd_Fault ; msg.topic = new MqttTopic(orgTag, com.protocol, com.rtuAddr, ProtocolConstantSdV1.PubTopicIrr) ; msg.msg = JSON.toJSONString(new StirStartVo(cvo.startTrueStopFalse ?(byte)1:0)) ; return msg ; } private MqttPubMsgSdV1 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") ; } MqttPubMsgSdV1 msg = new MqttPubMsgSdV1() ; this.setPubMsgBase(com, msg); msg.isCacheForOffLine = false ; msg.hasResponse = false ; msg.cd = CodeSdV1.cd_Param ; msg.topic = new MqttTopic(orgTag, com.protocol, com.rtuAddr, ProtocolConstantSdV1.PubTopicParam) ; msg.msg = JSON.toJSONString(new ParamSetVo(cvo.stirDuration, cvo.injectDuration)) ; return msg ; } private void setPubMsgBase(Command com, MqttPubMsgSdV1 msg){ msg.commandId = com.id ; msg.deviceId = com.rtuAddr ; msg.mqttResultSendWebUrl = com.rtuResultSendWebUrl ; } }