在remote子模块实现的”远程监测“功能（查询列表附在线状态）

liurunyu

2024-10-23 76b0ffff1d0d483b1837cb0d5fdbb59339210c3a

在remote子模块实现的”远程监测“功能（查询列表附在线状态）

 pipIrr-platform/pipIrr-global/src/main/java/com/dy/pipIrrGlobal/daoPr/PrIntakeMapper.java

@@ -127,6 +127,23 @@
    //List<VoOnLineIntake> getOnLineIntakes(@Param("onLineMap") String onLineMap, @Param("isOnLine") Boolean isOnLine);
    List<VoOnLineIntake> getOnLineIntakes(Map<?, ?> params);


    /**
     * 获取取水口数量（在线或不在线）
     * @param params
     * @return
     */
    Long selectIntakesCountForOnLine(Map<?, ?> params);

    /**
     * 获取取水口列表在线或不在线）
     * @param params
     * @return
     */
    List<VoOnLineIntake> selectIntakesForOnLine(Map<?, ?> params);



    /**
     * 根据取水口编号获取取水口对象
     * @param params

 pipIrr-platform/pipIrr-global/src/main/java/com/dy/pipIrrGlobal/rtuMw/CodeLocal.java

New file
@@ -0,0 +1,21 @@
package com.dy.pipIrrGlobal.rtuMw;

public class CodeLocal {

    public static final String clock = "LCD0000" ;//查询监控中间件时钟

    public static final String onLineAll = "LCD0001" ;//查询所有RTU在线情况

    public static final String onLinePart = "LCD0002" ;//查询部分RTU在线情况

    public static final String onLineStatistics = "LCD0003" ;//查询所有RTU状态情况

    public static final String allProtocols = "LCD0100" ;//查询所有协议配置

    public static final String stopTcpSv = "LCD0110" ;//停止TCP服务，不再接入新的TCP连接，已经TCP连接的全部断连接

    public static final String recoverTcpSv = "LCD0112" ;//重启TCP服务，接入新的TCP连接

    public static final String mwState = "LCD0200" ;//得到通信中间件运行信息

}

 pipIrr-platform/pipIrr-global/src/main/java/com/dy/pipIrrGlobal/rtuMw/ToRtuMwCom.java

New file
@@ -0,0 +1,85 @@
package com.dy.pipIrrGlobal.rtuMw;

import com.dy.common.multiDataSource.DataSourceContext;
import com.dy.common.mw.protocol.Command;
import com.dy.common.mw.protocol.CommandType;
import com.dy.common.webUtil.BaseResponse;
import org.springframework.core.env.Environment;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.ResponseEntity;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.util.UriComponentsBuilder;

/**
 * @Author: liurunyu
 * @Date: 2024/10/23 11:45
 * @Description
 */
public abstract class ToRtuMwCom {
    /**
     * pro_mw：属性
     * tag从控制器中获取
     * key_mw：url的key
     */
    private static final String pro_mw = "mw";
    private static final String key_mw = "comSendUrl";


    /**
     * 得到向通信中间件发送数据的URL
     * @param env
     * @return
     */
    protected String getToMwUrl(Environment env) {
        return env.getProperty(pro_mw + "." + DataSourceContext.get() + "." + key_mw);
    }

    /**
     * 创建外部命令（发给控制器）
     * @param code 命令code
     * @return
     */
    protected Command createOuterCommand(String comId, String code) {
        Command com = new Command();
        com.id = comId;
        com.code = code ;
        com.type = CommandType.outerCommand;
        return com ;
    }

    /**
     * 创建内部
     * @param code 命令code
     * @return
     */
    protected Command createInnerCommand(String code) {
        Command com = new Command();
        com.id = Command.defaultId;
        com.code = code ;
        com.type = CommandType.innerCommand;
        return com ;
    }

    /**
     * 发送命令
     *
     * @return
     */
    protected BaseResponse sendCom2Mw(RestTemplate restTemplate, String comSendUrl, Command com) {
        String url = UriComponentsBuilder.fromUriString(comSendUrl)
                .build()
                .toUriString();
        HttpHeaders headers = new HttpHeaders();
        HttpEntity<Command> httpEntity = new HttpEntity<>(com, headers);
        ResponseEntity<BaseResponse> response = null;
        try {
            // 通过Post方式调用接口
            response = restTemplate.exchange(url, HttpMethod.POST, httpEntity, BaseResponse.class);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return response.getBody();
    }
}

 pipIrr-platform/pipIrr-global/src/main/resources/mapper/PrIntakeMapper.xml

@@ -421,27 +421,27 @@
            LEFT JOIN pr_controller con ON con.intakeId = inta.id
            LEFT JOIN rm_on_hour_report_last hou ON hou.intake_id = inta.id
            LEFT JOIN(
            SELECT intake_id AS intakeId,
                   CONCAT(
                           IF(alarm_loss = 1, IF(alarm_water_meter_fault = 1, '流量计故障,', ''),
                              IF(alarm_water_meter_fault = 1, '流量计故障', '')),
                           IF(alarm_battery_volt = 1, IF(alarm_loss = 1, '漏损(偷水)报警,', ''),
                              IF(alarm_loss = 1, '漏损(偷水)报警', '')),
                           IF(alarm_valve = 1, IF(alarm_battery_volt = 1, '蓄电池电压报警,', ''),
                              IF(alarm_battery_volt = 1, '蓄电池电压报警', '')),
                           IF(alarm_valve = 1, '阀门报警', '')
                       )     AS alarm
            FROM rm_alarm_state_last
            WHERE (alarm_water_meter_fault = 1 OR alarm_loss = 1 OR alarm_battery_volt = 1 OR alarm_valve = 1)
              AND dt >= DATE_SUB(NOW(), INTERVAL 12 HOUR)
        ) alarm ON alarm.intakeId = inta.id
                SELECT intake_id AS intakeId,
                       CONCAT(
                               IF(alarm_loss = 1, IF(alarm_water_meter_fault = 1, '流量计故障,', ''),
                                  IF(alarm_water_meter_fault = 1, '流量计故障', '')),
                               IF(alarm_battery_volt = 1, IF(alarm_loss = 1, '漏损(偷水)报警,', ''),
                                  IF(alarm_loss = 1, '漏损(偷水)报警', '')),
                               IF(alarm_valve = 1, IF(alarm_battery_volt = 1, '蓄电池电压报警,', ''),
                                  IF(alarm_battery_volt = 1, '蓄电池电压报警', '')),
                               IF(alarm_valve = 1, '阀门报警', '')
                           )     AS alarm
                FROM rm_alarm_state_last
                WHERE (alarm_water_meter_fault = 1 OR alarm_loss = 1 OR alarm_battery_volt = 1 OR alarm_valve = 1)
                  AND dt >= DATE_SUB(NOW(), INTERVAL 12 HOUR)
            ) alarm ON alarm.intakeId = inta.id
            LEFT JOIN JSON_TABLE(
        <!--'[{"rtuAddr":"37142501020100215","isOnLine":true},{"rtuAddr":"4000004","isOnLine":true},{"rtuAddr":"dy20240325","isOnLine":false}]',-->
        #{onLineMap},
        '$[*]' COLUMNS (
            rtuAddr VARCHAR(20) PATH '$.rtuAddr',
            isOnLine BOOLEAN PATH '$.isOnLine'
            )
                <!--'[{"rtuAddr":"37142501020100215","isOnLine":true},{"rtuAddr":"4000004","isOnLine":true},{"rtuAddr":"dy20240325","isOnLine":false}]',-->
                #{onLineMap},
                '$[*]' COLUMNS (
                    rtuAddr VARCHAR(20) PATH '$.rtuAddr',
                    isOnLine BOOLEAN PATH '$.isOnLine'
                    )
            ) rtus ON con.rtuAddr = rtus.rtuAddr
        <where>
            <if test="isOnLine != null">
@@ -521,6 +521,97 @@
        </trim>
    </select>







    <!--获取取水口数量（在线或离线或不设状态）-->
    <select id="selectIntakesCountForOnLine" resultType="java.lang.Long">
        SELECT COUNT(*) AS recordCount
        FROM pr_intake inta
        INNER JOIN pr_controller con ON con.intakeId = inta.id
        <if test="onLineMap != null and onLineMap !='' and isOnLine != null">
            LEFT JOIN JSON_TABLE(
                <!--'[{"rtuAddr":"37142501020100215","isOnLine":true},{"rtuAddr":"4000004","isOnLine":true},{"rtuAddr":"dy20240325","isOnLine":false}]',-->
                #{onLineMap},
                '$[*]' COLUMNS (
                    rtuAddr VARCHAR(20) PATH '$.rtuAddr',
                    isOnLine BOOLEAN PATH '$.isOnLine'
                )
            ) rtus ON con.rtuAddr = rtus.rtuAddr
        </if>
        WHERE con.intakeId is not null
        <if test="isOnLine != null">
            AND rtus.isOnLine = #{isOnLine}
        </if>
        <if test="intakeNum != null and intakeNum != ''">
            AND inta.name = #{intakeNum}
        </if>
    </select>


    <!--获取取水口列表（在线或离线或不设状态）-->
    <select id="selectIntakesForOnLine" resultType="com.dy.pipIrrGlobal.voPr.VoOnLineIntake">
        SELECT
        inta.id                     AS intakeId,
        con.rtuAddr,
        inta.name                   AS intakeNum,
        inta.lng,
        inta.lat,
        IFNULL(hou.total_amount, 0) AS totalAmount,
        alarm.alarm
        FROM pr_intake inta
        INNER JOIN pr_controller con ON con.intakeId = inta.id
        LEFT JOIN rm_on_hour_report_last hou ON hou.intake_id = inta.id
        LEFT JOIN(
            SELECT intake_id AS intakeId,
            CONCAT(
                IF(alarm_loss = 1, '漏损报警，', ''),
                IF(alarm_battery_volt = 1, '电池电压报警，', ''),
                IF(alarm_valve = 1, '阀门报警，', ''),
                IF(alarm_water_meter_fault = 1, '流量计故障报警，', '')
            ) AS alarm
            FROM rm_alarm_state_last
            WHERE (alarm_loss = 1 OR alarm_battery_volt = 1 OR alarm_valve = 1 OR alarm_water_meter_fault = 1)
            AND dt >= DATE_SUB(NOW(), INTERVAL 12 HOUR)
        ) alarm ON alarm.intakeId = inta.id
        <if test="onLineMap != null and onLineMap !='' and isOnLine != null">
            LEFT JOIN JSON_TABLE(
                <!--'[{"rtuAddr":"37142501020100215","isOnLine":true},{"rtuAddr":"4000004","isOnLine":true},{"rtuAddr":"dy20240325","isOnLine":false}]',-->
                #{onLineMap},
                '$[*]' COLUMNS (
                    rtuAddr VARCHAR(20) PATH '$.rtuAddr',
                    isOnLine BOOLEAN PATH '$.isOnLine'
                )
            ) rtus ON con.rtuAddr = rtus.rtuAddr
        </if>
        WHERE con.intakeId is not null
        <if test="isOnLine != null">
            AND rtus.isOnLine = #{isOnLine}
        </if>
        <if test="intakeNum != null and intakeNum != ''">
            AND inta.name = #{intakeNum}
        </if>
        order by inta.id ASC
        <trim prefix="limit ">
            <if test="start != null and count != null">
                #{start,javaType=Integer,jdbcType=INTEGER}, #{count,javaType=Integer,jdbcType=INTEGER}
            </if>
        </trim>
    </select>











    <!--根据取水口编号获取取水口对象-->
    <select id="getIntakeByName" resultType="com.dy.pipIrrGlobal.voPr.VoOnLineIntake">
        SELECT con.intakeId,

 pipIrr-platform/pipIrr-web/pipIrr-web-remote/src/main/java/com/dy/pipIrrRemote/monitor/MonitorCtrl.java

New file
@@ -0,0 +1,49 @@
package com.dy.pipIrrRemote.monitor;

import com.dy.common.aop.SsoAop;
import com.dy.common.webUtil.BaseResponse;
import com.dy.common.webUtil.BaseResponseUtils;
import com.dy.common.webUtil.QueryResultVo;
import com.dy.pipIrrGlobal.voPr.VoOnLineIntake;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.List;

/**
 * @Author: liurunyu
 * @Date: 2024/10/23 11:32
 * @Description
 */

@Slf4j
@Tag(name = "监测控制", description = "监测控制")
@RestController
@RequestMapping(path = "monitor")
@RequiredArgsConstructor
public class MonitorCtrl {
    private final MonitorSv monitorSv;

    /**
     * 获取取水口列表（在线和不在线）
     *
     * @param qo
     * @return
     */
    @GetMapping(path = "all_intakes")
    @SsoAop()
    public BaseResponse<QueryResultVo<List<VoOnLineIntake>>> allIntakes(QueryVo qo) {
        try {
            QueryResultVo<List<VoOnLineIntake>> res = monitorSv.selectOnLineIntakes(qo);
            return BaseResponseUtils.buildSuccess(res);
        } catch (Exception e) {
            log.error("查询取水口异常", e);
            return BaseResponseUtils.buildException(e.getMessage());
        }
    }

}

 pipIrr-platform/pipIrr-web/pipIrr-web-remote/src/main/java/com/dy/pipIrrRemote/monitor/MonitorSv.java

New file
@@ -0,0 +1,174 @@
package com.dy.pipIrrRemote.monitor;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.dy.common.mw.protocol.Command;
import com.dy.common.mw.protocol.CommandBackParam;
import com.dy.common.webUtil.BaseResponse;
import com.dy.common.webUtil.QueryResultVo;
import com.dy.pipIrrGlobal.daoPr.PrIntakeMapper;
import com.dy.pipIrrGlobal.rtuMw.CodeLocal;
import com.dy.pipIrrGlobal.rtuMw.ToRtuMwCom;
import com.dy.pipIrrGlobal.voPr.VoOnLineIntake;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.common.utils.PojoUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @Author: liurunyu
 * @Date: 2024/10/23 11:32
 * @Description
 */
@Slf4j
@Service
public class MonitorSv extends ToRtuMwCom {

    @Autowired
    private PrIntakeMapper prIntakeMapper;

    @Autowired
    private Environment env;

    @Autowired
    private RestTemplate restTemplate;

    /**
     * 获取已经绑定控制器的取水口列表（并标记在线与不在线）
     *
     * @return
     */
    public QueryResultVo<List<VoOnLineIntake>> selectOnLineIntakes(QueryVo vo) {
        if(vo.getIsOnLine() == null){
            //不区分在线与离线
            return selectIntakesIgnoreOnOffLine(vo) ;
        }else{
            //在线 或 离线
            return selectIntakesOnOrOffLine(vo) ;
        }
    }

    /**
     * 查询取水口，不限制在线与离线状态
     * @param vo
     * @return
     */
    private QueryResultVo<List<VoOnLineIntake>> selectIntakesIgnoreOnOffLine(QueryVo vo) {
        QueryResultVo<List<VoOnLineIntake>> rsVo = this.queryDb(vo) ;
        if(rsVo != null && rsVo.obj != null && (rsVo.obj.size() > 0)){
            String rtuAddrs = "" ;
            boolean first = true ;
            for(VoOnLineIntake rVo : rsVo.obj){
                if(first){
                    first = false ;
                    rtuAddrs += rVo.getRtuAddr() ;
                }else{
                    rtuAddrs += "," + rVo.getRtuAddr() ;
                }
            }
            //向通信中间件发关命令，查询部分RTU在线情况
            Command com = this.createInnerCommand(CodeLocal.onLinePart);
            com.setParam(rtuAddrs) ;
            String comSendUrl = this.getToMwUrl(this.env) ;
            BaseResponse<Command> res = sendCom2Mw(restTemplate, comSendUrl, com) ;
            if(res != null){
                if(res.isSuccess()){
                    Command reCom = res.getContent() ;
                    CommandBackParam bakParam = JSON.parseObject(((JSONObject)reCom.param).toJSONString(), CommandBackParam.class) ;
                    if(bakParam != null){
                        if(bakParam.getSuccess().booleanValue()){
                            //通信中间件成功返回命令结果
                            HashMap<String, Boolean> onLineMap = JSON.parseObject(((JSONObject)reCom.getAttachment()).toJSONString(), HashMap.class);
                            for(VoOnLineIntake rVo : rsVo.obj){
                                if(onLineMap.containsKey(rVo.getRtuAddr())) {
                                    rVo.setIsOnLine(onLineMap.get(rVo.getRtuAddr()));
                                }
                            }
                        }
                    }else{
                        log.error("通信中间件返回内部命令结果中不包含CommandBackParam类型参数");
                    }
                }else{
                    log.error("通信中间件返回内部命令执行失败" + (res.getMsg() == null? "" : ("，" + res.getMsg()))) ;
                }
            }else{
                log.error("通信中间件返回内部命令结果为null");
            }
        }
        return rsVo;
    }


    /**
     * 查询取水口，在线或离线状态
     * @param vo
     * @return
     */
    private QueryResultVo<List<VoOnLineIntake>> selectIntakesOnOrOffLine(QueryVo vo) {
        //向通信中间件发关命令，查询部分RTU在线情况
        Command com = this.createInnerCommand(CodeLocal.onLineAll);
        String comSendUrl = this.getToMwUrl(this.env) ;
        BaseResponse<Command> res = sendCom2Mw(restTemplate, comSendUrl, com) ;
        if(res != null){
            if(res.isSuccess()){
                Command reCom = res.getContent() ;
                CommandBackParam bakParam = JSON.parseObject(((JSONObject)reCom.param).toJSONString(), CommandBackParam.class) ;
                if(bakParam != null){
                    if(bakParam.getSuccess().booleanValue()){
                        //通信中间件成功返回命令结果
                        HashMap<String, Boolean> onLineMap = JSON.parseObject(((JSONObject)reCom.getAttachment()).toJSONString(), HashMap.class);
                        JSONArray jsonArray = new JSONArray();
                        for (Map.Entry<String, Boolean> entry : onLineMap.entrySet()) {
                            JSONObject jsonObject = new JSONObject();
                            jsonObject.put("rtuAddr", entry.getKey());
                            jsonObject.put("isOnLine", entry.getValue());
                            jsonArray.add(jsonObject);
                        }
                        vo.setOnLineMap(jsonArray.toJSONString());
                    }
                }else{
                    log.error("通信中间件返回内部命令结果中不包含CommandBackParam类型参数");
                }
            }else{
                log.error("通信中间件返回内部命令执行失败" + (res.getMsg() == null? "" : ("，" + res.getMsg()))) ;
            }
        }else{
            log.error("通信中间件返回内部命令结果为null");
        }

        return this.queryDb(vo) ;
    }

    /**
     * 查询数据库
     * @param vo
     * @return
     */
    private QueryResultVo<List<VoOnLineIntake>> queryDb(QueryVo vo){
        Map<String, Object> params = (Map<String, Object>) PojoUtils.generalize(vo);
        Long itemTotal = prIntakeMapper.selectIntakesCountForOnLine(params);

        QueryResultVo<List<VoOnLineIntake>> rsVo = new QueryResultVo<>();
        rsVo.pageSize = vo.pageSize;
        rsVo.pageCurr = vo.pageCurr;
        rsVo.calculateAndSet(itemTotal, params);
        List<VoOnLineIntake> records = prIntakeMapper.selectIntakesForOnLine(params) ;
        rsVo.obj = records ;
        for(VoOnLineIntake rVo : records){
            if(rVo.getAlarm() != null && !rVo.getAlarm().trim().equals("") && rVo.getAlarm().endsWith("，")){
                rVo.setAlarm(rVo.getAlarm().substring(0, rVo.getAlarm().length() - 1)) ;
            }
        }
        return rsVo ;
    }



}

 pipIrr-platform/pipIrr-web/pipIrr-web-remote/src/main/java/com/dy/pipIrrRemote/monitor/QueryVo.java

New file
@@ -0,0 +1,32 @@
package com.dy.pipIrrRemote.monitor;

import com.dy.common.webUtil.QueryConditionVo;
import lombok.Data;

/**
 * @author ZhuBaoMin
 * @date 2024-05-27 20:32
 * @LastEditTime 2024-05-27 20:32
 * @Description
 */

@Data
public class QueryVo extends QueryConditionVo {
    /**
     * 取水口编号
     */
    private String intakeNum;

    /**
     * 是否在线
     */

    private Boolean isOnLine;


    /**
     * 非前端设置的参数据，中间件返回的RTU在线情况对象数组
     */
    private String onLineMap;

}