任务提交 mqtt 完成全部任务上报

ruoyi-system/src/main/java/com/ruoyi/system/domain/BerthingFloatConfig.java

@@ -15,7 +15,12 @@ import lombok.Data;
 @Data
 public class BerthingFloatConfig implements Serializable {
     private static final long serialVersionUID = -49994754277835773L;
-/**
+
+    /**
+     * 设备类型，0：移动测流；1：固定测流
+     */
+    private Long type;
+    /**
      * 水位变幅ID
      */
     private Long floatId;

ruoyi-system/src/main/java/com/ruoyi/system/dto/BerthingFloatDTO.java

@@ -19,6 +19,13 @@ public class BerthingFloatDTO implements Serializable {
      */
     private Long floatId;
 
+
+    /**
+     * 设备类型，0：移动测流；1：固定测流
+     */
+    @NotNull(message = "设备类型不能为空")
+    private Long type;
+
 /**
      * 站点ID
      */

ruoyi-system/src/main/java/com/ruoyi/system/dto/ManualMeasureDTO.java

@@ -22,6 +22,7 @@ public class ManualMeasureDTO implements Serializable {
      */
     @NotNull(message = "测量类型不能为空")
     private Long type;
+
     /**
      * 停泊点数组(当type为2时有效)，当type为1时改项为null 断面点位json list字符串 例子：[{“x”:54.3,"y":44.2}]
      */

ruoyi-system/src/main/java/com/ruoyi/system/dto/SiteConfigMqttDTO.java

@@ -0,0 +1,128 @@
+package com.ruoyi.system.dto;
+
+import lombok.Data;
+
+import javax.validation.constraints.NotBlank;
+import javax.validation.constraints.NotNull;
+import java.io.Serializable;
+
+/**
+ * 站点配置信息(SiteConfig)实体类
+ *
+ * @author makejava
+ * @since 2024-04-09 14:14:59
+ */
+@Data
+public class SiteConfigMqttDTO implements Serializable {
+
+/**
+     * 设备类型，0：移动测流；1：固定测流
+     */
+    private Integer type;
+/**
+     * 站码
+     */
+    private String id;
+/**
+     * 左岸右岸标记（0为右岸，1为左岸）
+     */
+    private Integer local;
+/**
+     * 小车位置的偏移量
+     */
+    private Double offset;
+/**
+     * 小车测流停泊时间 s
+     */
+    private Long stoptime;
+/**
+     * 水位采集时间间隔 s
+     */
+    private Long wltime;
+/**
+     * 低、中、高 行驶速度，1，2，3
+     */
+    private Integer speed;
+/**
+     * 测流的最低水位
+     */
+    private Double wlevelmin;
+/**
+     * 测流的最高水位
+     */
+    private Double wlevelmax;
+/**
+     * 测流低电量保护（百分比）
+     */
+    private Long eqmin;
+/**
+     * 水位基值
+     */
+    private Double wbaselevel;
+/**
+     * 水位计485从机地址
+     */
+    private Long wladdr;
+/**
+     * 水位计数据类型，0为int型1为float型
+     */
+    private Integer wltype;
+/**
+     * 水位计存储首地址
+     */
+    private Long wlfirstadd;
+/**
+     * 水位计缩放精度
+     */
+    private Long wlzoom;
+/**
+     * 水位计数据长度
+     */
+    private Long wllength;
+/**
+     * 水位计度量单位 0cm  1m
+     */
+    private Integer wlunit;
+/**
+     * 水位计偏移量
+     */
+    private Long wloffset;
+/**
+     * 水位修正系数，为1时不修正
+     */
+    private Double wlfactor;
+/**
+     * 流速仪485从机地址
+     */
+    private Long wsaddr;
+/**
+     * 流速仪数据类型，0为int型1为float型
+     */
+    private Integer wstype;
+/**
+     * 流速仪存储首地址
+     */
+    private Long wsfirstadd;
+/**
+     * 流速仪缩放精度
+     */
+    private Long wszoom;
+/**
+     * 流速仪数据长度
+     */
+    private Long wslength;
+/**
+     * 流速仪度量单位 0cm  1m
+     */
+    private Integer wsunit;
+/**
+     * 流速仪偏移量
+     */
+    private Long wsoffset;
+/**
+     * 流速修正系数，为1时不修正
+     */
+    private Double wsfactor;
+
+}
+

ruoyi-system/src/main/java/com/ruoyi/system/service/DirectiveService.java

@@ -21,6 +21,6 @@ public interface DirectiveService {
 
     boolean manualMeasure(ManualMeasureDTO manualMeasureDTO);
 
-    boolean taskStop(Long siteId);
+    boolean taskStop(Long siteId, Long action);
 
 }

ruoyi-system/src/main/java/com/ruoyi/system/service/impl/BerthingServiceImpl.java

@@ -246,11 +246,20 @@ public class BerthingServiceImpl implements BerthingService {
                 berthingPointConfig.setFactors(JSONUtil.toJsonStr(berthingPointConfigDTO.getFactors()));
                 res =  berthingPointConfigMapper.update(berthingPointConfig) > 0;
             }
+            //TODO 下发配置
+            String topic = "down/" + siteInfo.getDeviceId() + "/stopPosition";
+            HashMap<String, Object> requestMap = new HashMap<>();
+            requestMap.put("id",siteInfo.getId());
+            requestMap.put("wlevel",berthingPointConfig.getWlevel());
+            requestMap.put("planid",berthingPointConfig.getPlanid());
+            requestMap.put("positions",positions);
+            requestMap.put("factors",berthingPointConfigDTO.getFactors());
+            String data = JSONUtil.toJsonStr(requestMap);
+            mqttGateWayService.sendMessageToMqtt(data, topic);
         } catch (Exception e){
             log.error("新增或修改停泊点错误", e);
             throw new BaseException("新增或修改停泊点错误");
         }
-        //TODO 下发配置
         return res;
 }
 
@@ -315,11 +324,20 @@ public class BerthingServiceImpl implements BerthingService {
                 berthingTime.setType(berthingTimeDTO.getType());
                 res =  berthingTimeConfigMapper.update(berthingTime) > 0;
             }
+            //TODO 下发配置
+            String topic = "down/" + siteInfo.getDeviceId() + "/measureMode";
+            HashMap<String, Object> requestMap = new HashMap<>();
+            requestMap.put("id",siteInfo.getId());
+            requestMap.put("type",berthingTime.getType());
+            requestMap.put("planid",berthingTime.getPlanid());
+            requestMap.put("wlevel",berthingTime.getWlevel());
+            requestMap.put("times",times);
+            String data = JSONUtil.toJsonStr(requestMap);
+            mqttGateWayService.sendMessageToMqtt(data, topic);
         } catch (Exception e){
             log.error("新增或修改时间策略错误", e);
             throw new BaseException("新增或修改时间策略错误");
         }
-        //TODO 下发配置
         return res;
     }
 
@@ -374,20 +392,32 @@ public class BerthingServiceImpl implements BerthingService {
                 berthingFloat.setId(siteInfo.getId());
                 berthingFloat.setTimespan(berthingFloatDTO.getTimespan());
                 berthingFloat.setWlevelchange(berthingFloatDTO.getWlevelchange());
+                berthingFloat.setType(berthingFloatDTO.getType());
                 res =  berthingFloatConfigMapper.insert(berthingFloat) > 0;
             } else {
                 //更新
+                berthingFloat.setType(berthingFloatDTO.getType());
                 berthingFloat.setPlanid(berthingFloatDTO.getPlanid());
                 berthingFloat.setWlevel(berthingFloatDTO.getWlevel());
                 berthingFloat.setTimespan(berthingFloatDTO.getTimespan());
                 berthingFloat.setWlevelchange(berthingFloatDTO.getWlevelchange());
                 res =  berthingFloatConfigMapper.update(berthingFloat) > 0;
             }
+            //TODO 下发配置
+            String topic = "down/" + siteInfo.getDeviceId() + "/addMode";
+            HashMap<String, Object> requestMap = new HashMap<>();
+            requestMap.put("id",siteInfo.getId());
+            requestMap.put("type",berthingFloat.getType());
+            requestMap.put("planid",berthingFloat.getPlanid());
+            requestMap.put("wlevel",berthingFloat.getWlevel());
+            requestMap.put("timespan",berthingFloat.getTimespan());
+            requestMap.put("wlevelchange",berthingFloat.getWlevelchange());
+            String data = JSONUtil.toJsonStr(requestMap);
+            mqttGateWayService.sendMessageToMqtt(data, topic);
         } catch (Exception e){
             log.error("新增或修改水位变幅策略错误", e);
             throw new BaseException("新增或修改水位变幅策略错误");
         }
-        //TODO 下发配置
         return res;
     }
 

ruoyi-system/src/main/java/com/ruoyi/system/service/impl/DirectiveServiceImpl.java

@@ -1,6 +1,7 @@
 package com.ruoyi.system.service.impl;
 
 
+import cn.hutool.json.JSONUtil;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import com.ruoyi.common.exception.base.BaseException;
 import com.ruoyi.system.domain.SiteConfig;
@@ -12,6 +13,7 @@ import com.ruoyi.system.mapper.SiteConfigMapper;
 import com.ruoyi.system.mapper.SiteInfoMapper;
 import com.ruoyi.system.paramet.SiteInfoQuery;
 import com.ruoyi.system.service.DirectiveService;
+import com.ruoyi.system.service.MqttGateWayService;
 import com.ruoyi.system.service.SiteService;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.BeanUtils;
@@ -20,6 +22,7 @@ import org.springframework.transaction.annotation.Transactional;
 
 import javax.annotation.Resource;
 import java.util.Date;
+import java.util.HashMap;
 
 
 /**
@@ -33,6 +36,9 @@ public class DirectiveServiceImpl implements DirectiveService {
     @Resource
     private SiteInfoMapper siteInfoMapper;
 
+    @Resource
+    MqttGateWayService mqttGateWayService;
+
     @Override
     public boolean manualMeasure(ManualMeasureDTO manualMeasureDTO) {
         SiteInfo siteInfo = siteInfoMapper.queryById(manualMeasureDTO.getSiteId());
@@ -40,8 +46,19 @@ public class DirectiveServiceImpl implements DirectiveService {
             throw new BaseException("站点不存在");
         }
         try {
-            log.info("手动测量信息下发{}", manualMeasureDTO);
+            log.info("手动下发任务 SiteId:{}", manualMeasureDTO.getSiteId());
+            HashMap<String, Object> requestMap = new HashMap<>();
+            requestMap.put("id",siteInfo.getId());
+            requestMap.put("type",manualMeasureDTO.getType());
+            if (manualMeasureDTO.getType() == 1){
+                requestMap.put("positions",null);
+            } else {
+                requestMap.put("positions",manualMeasureDTO.getPositions());
+            }
             //TODO 发送mqtt
+            String topic = "down/" + siteInfo.getDeviceId() + "/manualMeasure";
+            String data = JSONUtil.toJsonStr(requestMap);
+            mqttGateWayService.sendMessageToMqtt(data, topic);
         } catch (Exception e){
             log.error("手动测量下发异常", e);
             throw new BaseException("手动测量下发异常");
@@ -51,14 +68,20 @@ public class DirectiveServiceImpl implements DirectiveService {
 
 
     @Override
-    public boolean taskStop(Long siteId) {
+    public boolean taskStop(Long siteId,Long action) {
         SiteInfo siteInfo = siteInfoMapper.queryById(siteId);
         if (siteInfo == null) {
             throw new BaseException("站点不存在");
         }
         try {
-            log.info("手动停止任务 {}", siteId);
+            log.info("手动停止任务 SiteId:{} action:{}", siteId,action);
             //TODO 发送mqtt
+            HashMap<String, Object> requestMap = new HashMap<>();
+            requestMap.put("id",siteInfo.getId());
+            requestMap.put("action",action);
+            String topic = "down/" + siteInfo.getDeviceId() + "/taskStop";
+            String data = JSONUtil.toJsonStr(requestMap);
+            mqttGateWayService.sendMessageToMqtt(data, topic);
         } catch (Exception e){
             log.error("手动测量下发异常", e);
             throw new BaseException("手动测量下发异常");

ruoyi-system/src/main/java/com/ruoyi/system/service/impl/SiteServiceImpl.java

@@ -2,16 +2,17 @@ package com.ruoyi.system.service.impl;
 
 
 import cn.hutool.json.JSONUtil;
-import com.alibaba.fastjson2.JSON;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import com.ruoyi.common.exception.base.BaseException;
 import com.ruoyi.system.domain.SiteConfig;
 import com.ruoyi.system.domain.SiteInfo;
 import com.ruoyi.system.dto.SiteConfigDTO;
+import com.ruoyi.system.dto.SiteConfigMqttDTO;
 import com.ruoyi.system.dto.SiteInfoDTO;
 import com.ruoyi.system.mapper.SiteConfigMapper;
 import com.ruoyi.system.mapper.SiteInfoMapper;
 import com.ruoyi.system.paramet.SiteInfoQuery;
+import com.ruoyi.system.service.MqttGateWayService;
 import com.ruoyi.system.service.SiteService;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.BeanUtils;
@@ -36,6 +37,9 @@ public class SiteServiceImpl implements SiteService {
     @Resource
     private SiteConfigMapper siteConfigMapper;
 
+    @Resource
+    private MqttGateWayService mqttGateWayService;
+
     @Transactional(rollbackFor = Exception.class)
     @Override
     public boolean insert(SiteInfoDTO siteInfoDTO) {
@@ -125,16 +129,10 @@ public class SiteServiceImpl implements SiteService {
             boolean res = siteConfigMapper.insert(siteConfig) > 0;
             //TODO 下发MQTT消息
             String topic = "down/" + siteInfo.getDeviceId() + "/settings";
-            // 配置 SerializeConfig，开启过滤 null 值的特性
-//            SerializeConfig serializeConfig = SerializeConfig.globalInstance;
-//            serializeConfig.configForBean().setNullValueFilter((k, v) -> v != null);
-//
-//            // 将 User 对象转换为 JSON 字符串，过滤 null 值
-//            String data = JSON.toJSONString(siteConfig, serializeConfig, SerializerFeature.WriteMapNullValue,
-//                    SerializerFeature.WriteNullStringAsEmpty, SerializerFeature.WriteNullListAsEmpty);
-//
-//            String data = JSONUtil.toJsonStr(siteConfig);
-//            mqttGateWayService.sendMessageToMqtt(data, topic);
+            SiteConfigMqttDTO siteConfigMqttDTO = new SiteConfigMqttDTO();
+            BeanUtils.copyProperties(siteConfig, siteConfigMqttDTO);
+            String data = JSONUtil.toJsonStr(siteConfigMqttDTO);
+            mqttGateWayService.sendMessageToMqtt(data, topic);
             return res;
         } catch (BaseException e){
             throw new BaseException(e.getMessage());
@@ -147,7 +145,14 @@ public class SiteServiceImpl implements SiteService {
     @Override
     public SiteConfig queryConfig(Long SiteId) {
         try {
-            return siteConfigMapper.queryBySiteId(SiteId);
+            SiteConfig siteConfig = siteConfigMapper.queryBySiteId(SiteId);
+//            SiteInfo siteInfo = siteInfoMapper.queryById(SiteId);
+//            String topic = "down/" + siteInfo.getDeviceId() + "/settings";
+//            SiteConfigMqttDTO siteConfigMqttDTO = new SiteConfigMqttDTO();
+//            BeanUtils.copyProperties(siteConfig, siteConfigMqttDTO);
+//            String data = JSONUtil.toJsonStr(siteConfigMqttDTO);
+//            mqttGateWayService.sendMessageToMqtt(data, topic);
+            return siteConfig;
         } catch (Exception e){
             log.error("查询站点信息错误", e);
             throw new BaseException("查询站点信息错误");
@@ -161,6 +166,10 @@ public class SiteServiceImpl implements SiteService {
             if (siteConfigDTO.getSiteId() == null) {
                 throw new BaseException("站点id不能为空");
             }
+            SiteInfo siteInfo = siteInfoMapper.queryById(siteConfigDTO.getSiteId());
+            if (siteInfo == null){
+                throw new BaseException("新增失败，SiteId错误");
+            }
             SiteConfig siteConfig;
             siteConfig = siteConfigMapper.queryBySiteId(siteConfigDTO.getSiteId());
             if (siteConfig == null) {
@@ -170,8 +179,14 @@ public class SiteServiceImpl implements SiteService {
             siteConfig = new SiteConfig();
             BeanUtils.copyProperties(siteConfigDTO, siteConfig);
             siteConfig.setConfigId(configId);
-            return siteConfigMapper.update(siteConfig) > 0;
-            //TODO 下发MQTT消息
+            boolean res = siteConfigMapper.update(siteConfig) > 0;
+            //TODO 下发MQTT消息测试
+            String topic = "down/" + siteInfo.getDeviceId() + "/settings";
+            SiteConfigMqttDTO siteConfigMqttDTO = new SiteConfigMqttDTO();
+            BeanUtils.copyProperties(siteConfig, siteConfigMqttDTO);
+            String data = JSONUtil.toJsonStr(siteConfigMqttDTO);
+            mqttGateWayService.sendMessageToMqtt(data, topic);
+            return res;
         } catch (BaseException e){
             throw new BaseException(e.getMessage());
         } catch (Exception e){

ruoyi-system/src/main/resources/mapper/BerthingFloatConfigMapper.xml

@@ -14,12 +14,13 @@
         <result property="timespan" column="timespan" jdbcType="INTEGER"/>
         <result property="wlevelchange" column="wlevelchange" jdbcType="NUMERIC"/>
         <result property="siteName" column="site_name" jdbcType="VARCHAR"/>
+        <result property="type" column="type" jdbcType="INTEGER"/>
     </resultMap>
 
     <!--查询单个-->
     <select id="queryById" resultMap="BerthingFloatConfigMap">
         select
-float_id, id, site_id, planid, create_time, status, is_del, wlevel, timespan, wlevelchange
+            float_id, id, site_id, planid, create_time, status, is_del, wlevel, timespan, wlevelchange, type
         from berthing_float_config
         where float_id = #{floatId}
     </select>
@@ -27,7 +28,7 @@ float_id, id, site_id, planid, create_time, status, is_del, wlevel, timespan, wl
     <!--查询指定行数据-->
     <select id="queryAllByLimit" resultMap="BerthingFloatConfigMap">
         select
-            a.float_id, a.id, a.site_id, a.planid, a.create_time, a.status, a.is_del, a.wlevel, a.timespan, a.wlevelchange, b.site_name
+            a.float_id, a.id, a.site_id, a.planid, a.create_time, a.status, a.is_del, a.wlevel, a.timespan, a.wlevelchange, a.type, b.site_name
         from berthing_float_config as a
         join site_info as b on a.site_id = b.site_id
         <where>
@@ -76,28 +77,31 @@ float_id, id, site_id, planid, create_time, status, is_del, wlevel, timespan, wl
             <if test="wlevelchange != null">
                 and wlevelchange = #{wlevelchange}
             </if>
+            <if test="type != null">
+                and type = #{type}
+            </if>
         </where>
     </select>
 
     <!--新增所有列-->
     <insert id="insert" keyProperty="floatId" useGeneratedKeys="true">
-        insert into berthing_float_config(id, site_id, planid, create_time, status, is_del, wlevel, timespan, wlevelchange)
-        values (#{id}, #{siteId}, #{planid}, #{createTime}, #{status}, #{isDel}, #{wlevel}, #{timespan}, #{wlevelchange})
+        insert into berthing_float_config(id, site_id, planid, create_time, status, is_del, wlevel, timespan, wlevelchange, type)
+        values (#{id}, #{siteId}, #{planid}, #{createTime}, #{status}, #{isDel}, #{wlevel}, #{timespan}, #{wlevelchange}, #{type})
     </insert>
 
     <insert id="insertBatch" keyProperty="floatId" useGeneratedKeys="true">
-        insert into berthing_float_config(id, site_id, planid, create_time, status, is_del, wlevel, timespan, wlevelchange)
+        insert into berthing_float_config(id, site_id, planid, create_time, status, is_del, wlevel, timespan, wlevelchange, type)
         values
         <foreach collection="entities" item="entity" separator=",">
-        (#{entity.id}, #{entity.siteId}, #{entity.planid}, #{entity.createTime}, #{entity.status}, #{entity.isDel}, #{entity.wlevel}, #{entity.timespan}, #{entity.wlevelchange})
+        (#{entity.id}, #{entity.siteId}, #{entity.planid}, #{entity.createTime}, #{entity.status}, #{entity.isDel}, #{entity.wlevel}, #{entity.timespan}, #{entity.wlevelchange}, #{entity.type})
         </foreach>
     </insert>
 
     <insert id="insertOrUpdateBatch" keyProperty="floatId" useGeneratedKeys="true">
-        insert into berthing_float_config(id, site_id, planid, create_time, status, is_del, wlevel, timespan, wlevelchange)
+        insert into berthing_float_config(id, site_id, planid, create_time, status, is_del, wlevel, timespan, wlevelchange, type)
         values
         <foreach collection="entities" item="entity" separator=",">
-            (#{entity.id}, #{entity.siteId}, #{entity.planid}, #{entity.createTime}, #{entity.status}, #{entity.isDel}, #{entity.wlevel}, #{entity.timespan}, #{entity.wlevelchange})
+            (#{entity.id}, #{entity.siteId}, #{entity.planid}, #{entity.createTime}, #{entity.status}, #{entity.isDel}, #{entity.wlevel}, #{entity.timespan}, #{entity.wlevelchange}, #{entity.type})
         </foreach>
         on duplicate key update
 id = values(id),
@@ -108,7 +112,8 @@ status = values(status),
 is_del = values(is_del),
 wlevel = values(wlevel),
 timespan = values(timespan),
-wlevelchange = values(wlevelchange)
+wlevelchange = values(wlevelchange),
+type = values(type)
     </insert>
 
     <!--通过主键修改数据-->
@@ -142,6 +147,9 @@ wlevelchange = values(wlevelchange)
             <if test="wlevelchange != null">
                 wlevelchange = #{wlevelchange},
             </if>
+            <if test="type != null">
+                type = #{type},
+            </if>
         </set>
         where float_id = #{floatId}
     </update>

waterAffairs-admin/src/main/java/com/ruoyi/web/controller/tool/DirectiveController.java

@@ -35,9 +35,9 @@ public class DirectiveController {
     }
 
     @PostMapping(value = "/taskStop")
-    @ApiOperation("任务停止")
-    public R<Boolean> taskStop(Long siteId)  {
-        return R.ok(directiveService.taskStop(siteId));
+    @ApiOperation("任务停止 action 0：终止 1：暂停 2：继续")
+    public R<Boolean> taskStop(Long siteId, Long action)  {
+        return R.ok(directiveService.taskStop(siteId,action));
     }