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 ; } }