游仙分库协议提交，待调整

jiazx0107@163.com

2024-01-05 7fe5e6110fb2516a81c2258e8d29fc19ee781d76

 游仙分库协议提交，待调整

 src/main/java/com/fzzy/protocol/ProtocolScheduled.java

@@ -80,6 +80,4 @@

        }
    }


}

 src/main/java/com/fzzy/protocol/youxian1/ServiceUtils.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,9 @@
package com.fzzy.protocol.youxian1;


/**
 * å·¥å…·ç±»
 */
public class ServiceUtils {

}

 src/main/java/com/fzzy/protocol/youxian1/analysis/AnalysisService.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,360 @@
package com.fzzy.protocol.youxian1.analysis;

import com.alibaba.fastjson.JSONObject;
import com.fzzy.api.data.GatewayDeviceType;
import com.fzzy.api.utils.BytesUtil;
import com.fzzy.api.utils.NumberUtil;
import com.fzzy.gateway.GatewayUtils;
import com.fzzy.gateway.api.GatewayDeviceReportService;
import com.fzzy.gateway.api.GatewayRemoteManager;
import com.fzzy.gateway.data.BaseReqData;
import com.fzzy.gateway.data.WeatherWebDto;
import com.fzzy.gateway.entity.GatewayDevice;
import com.fzzy.gateway.hx2023.ScConstant;
import com.fzzy.gateway.hx2023.data.*;
import com.fzzy.protocol.ProtocolUtils;
import com.fzzy.protocol.bhzn.cmd.ReMessageBuilder;
import com.fzzy.protocol.data.THDto;
import com.fzzy.protocol.youxian0.ServiceUtils;
import com.fzzy.protocol.youxian0.data.GrainRoot;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import javax.persistence.criteria.CriteriaBuilder;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * æ¸¸ä»™åˆ†åº“，返回报文解析
 */
@Slf4j
@Component(AnalysisService.BEAN_ID)
public class AnalysisService {

    public static final String BEAN_ID = "youxian1.analysisService";

    @Resource
    private GatewayRemoteManager gatewayRemoteManager;

    /**
     * ç”¨äºŽå­˜æ”¾è¿”回的仓温仓湿信息
     */
    public static Map<String, THDto> contextMapTH = new HashMap<>();


    /**
     * åè®®è§£æž
     * <p>
     * 80 80 80 80 80 90 41   --90代表分机编码
     * 39 39 39 35 39 39 39 39 39 39 39 39  --四组湿度
     * B4 30 30 31 30 30 B4 30 30 B4 30 30 --四组温度
     * ---以下为粮温，3个值转换为一个点
     * 31 30 35 31 37 32 32 31 34 32 30 33 31 30 38 31 36 34 32 32 35 32 31 38 31 30 34 31 36 36 32 32 33 32 31 34 31 30 32 31 35 39 32 31 37 32 30 34 31 30 37 31 33 37 32 32 34 32 31 37 31 31 30
     *
     * @param address
     * @param port
     * @param strMsg
     */
    public void analysis(InetAddress address, int port, String strMsg) {

        //分机ID
        int start = 5 * 2;
        int end = start + 1 * 2;
        String deviceSn = strMsg.substring(start, end);
        // deviceSn =  è°ƒæ•´è½¬æ¢è§„则 ---TODO

        GatewayDevice device = GatewayUtils.getCacheByDeviceSn(deviceSn);


        //解析温湿度
        this.analysisGrainTh(device, strMsg);


        //解析粮情
        this.analysisGrainStep1(device, strMsg);

    }

    private void analysisGrainTh(GatewayDevice device, String strMsg) {
        //湿度信息，四组，获取有效的一组数据
        int start = 7 * 2;
        int end = start + 12 * 2;
        String hHex = strMsg.substring(start, end);

        THDto thDto = new THDto();
        double h = getGrainTemp(hHex.substring(0, 6));
        if (h < 99.9) {
            thDto.setHumidityIn(h);
        }
        h = getGrainTemp(hHex.substring(6, 12));
        if (h < 99.9) {
            thDto.setHumidityIn(h);
        }
        h = getGrainTemp(hHex.substring(12, 18));
        if (h < 99.9) {
            thDto.setHumidityIn(h);
        }
        h = getGrainTemp(hHex.substring(18, 24));
        if (h < 99.9) {
            thDto.setHumidityIn(h);
        }

        //湿度信息
        start = 19 * 2;
        end = start + 12 * 2;
        String tHex = strMsg.substring(start, end);
        double t = getGrainTemp(tHex.substring(0, 6));
        if (t > -40.0) {
            thDto.setTempIn(t);
        }
        t = getGrainTemp(tHex.substring(6, 12));
        if (t > -40.0) {
            thDto.setTempIn(t);
        }
        t = getGrainTemp(tHex.substring(12, 18));
        if (t > -40.0) {
            thDto.setTempIn(t);
        }
        t = getGrainTemp(tHex.substring(18, 24));
        if (t > -40.0) {
            thDto.setTempIn(t);
        }

        this.add2Map(device.getDeviceSn(), thDto);
    }


    /**
     * åè®®è§£æž
     * <p>
     * 80 80 80 80 80 90 41   --90代表分机编码
     * 39 39 39 35 39 39 39 39 39 39 39 39  --四组湿度
     * B4 30 30 31 30 30 B4 30 30 B4 30 30 --四组温度
     * ---以下为粮温，3个值转换为一个点
     * 31 30 35 31 37 32 32 31 34 32 30 33 31 30 38 31 36 34 32 32 35 32 31 38 31 30 34 31 36 36 32 32 33 32 31 34 31 30 32 31 35 39 32 31 37 32 30 34 31 30 37 31 33 37 32 32 34 32 31 37 31 31 30
     *
     * @param device
     * @Param strMsg
     */
    private void analysisGrainStep1(GatewayDevice device, String strMsg) {

        String[] attCable = device.getCableRule().split("-");
        int cableZ = Integer.valueOf(attCable[0]);
        int cableY = Integer.valueOf(attCable[1]);
        int cableX = Integer.valueOf(attCable[2]);
        log.info("z={},x={},y={}", cableZ, cableX, cableY);

        //获取请求信息
        BaseReqData reqData = ProtocolUtils.getSyncReq(device.getDepotIdSys());
        if (null == reqData) {
            log.error("---------没有获取到请求信息，不执行解析------{}", device.getDeviceName());
            return;
        }

        //只保留粮情信息
        int start = 31 * 2;
        strMsg = strMsg.substring(start);
        int num = cableZ * cableY * cableX;
        String tempStr = "";
        List<Double> points = new ArrayList<>();
        double point = 0;
        for (int i = 0; i < num; i++) {
            tempStr = strMsg.substring(i * 3 * 2, i * 3 * 2 + 3 * 2);
            point = this.getGrainTemp(tempStr);

            //处理无效的数据
            if (point > 50) {
                point = -100.0;
            }

            points.add(point);
        }


        analysisGrainStep2(reqData, cableZ, cableY, cableX, points);
    }


    /**
     * æ•°æ®è½¬æ¢
     *
     * @param reqData
     * @param cableZ
     * @param cableY
     * @param cableX
     * @param points
     */
    private void analysisGrainStep2(BaseReqData reqData, int cableZ, int cableY, int cableX, List<Double> points) {

        GatewayDevice device = reqData.getDevice();
        //数据封装
        GrainData grain = new GrainData();
        grain.setMessageId(ScConstant.getMessageId());
        grain.setDeviceId(device.getDeviceId());
        grain.setTimestamp(System.currentTimeMillis() + "");

        ClientHeaders headers = new ClientHeaders();
        headers.setDeviceName(device.getDeviceName());
        headers.setProductId(device.getProductId());
        headers.setOrgId(device.getOrgId());
        headers.setMsgId(reqData.getMessageId());
        grain.setHeaders(headers);


        GrainOutPut outPut = new GrainOutPut();


        double max = ReMessageBuilder.MAX_TEMP, min = ReMessageBuilder.MIN_TEMP, sumT = 0.0, sumNum = cableX * cableY * cableZ;

        List<GrainTemp> temperature = new ArrayList<>();
        //根号
        int cableNum = 1, position = 0;

        double curTemp;
        int x = 0, y = 0, z = 0;
        for (int i = 0; i < points.size(); i++) {
            curTemp = points.get(i);
            position = i;
            z = i % cableZ + 1;
            x = i / (cableZ * cableY);
            y = x * (cableZ * cableY);
            y = (i - y) / cableZ;
            //根号
            cableNum = (i / cableZ) + 1;

            temperature.add(new GrainTemp(cableNum + "", z + "", curTemp + "", position + ""));

            sumT += curTemp;
            if (curTemp > max) {
                max = curTemp;
            }
            if (curTemp < min) {
                min = curTemp;
            }
        }

        if (sumNum == 0) {
            sumNum = 1;
            log.warn("---当前粮情采集异常--");
        }
        //过滤比较用的最大最小值
        if (max == ReMessageBuilder.MAX_TEMP) {
            max = 0.0;
        }
        if (min == ReMessageBuilder.MIN_TEMP) {
            min = 0.0;
        }

        outPut.setTemperature(temperature);
        outPut.setAvgTemperature(NumberUtil.keepPrecision((sumT / sumNum), 1) + "");
        outPut.setMinTemperature(min + "");
        outPut.setMaxTemperature(min + "");
        List<GrainTH> ths = new ArrayList<>();


        //获取温湿度
        THDto thDto = this.getTH(device.getDeviceSn());

        ths.add(new GrainTH(thDto.getTempIn() != null ? thDto.getTempIn() + "" : "", thDto.getHumidityIn() != null ? thDto.getHumidityIn() + "" : "", "1"));
        outPut.setTemperatureAndhumidity(ths);
        grain.setOutput(JSONObject.toJSONString(outPut));

        GatewayDevice gatewayDeviceWeather = GatewayUtils.getCacheByDeviceTypeOne(GatewayDeviceType.TYPE_09.getCode());

        //系统气象站信息
        WeatherWebDto weather = WeatherWebDto.contextMap.get("default");

        //气象信息
        GrainWeather weatherStation = new GrainWeather();
        weatherStation.setMessageId(ScConstant.getMessageId());
        weatherStation.setMessgeId(weatherStation.getMessageId());

        if (null != gatewayDeviceWeather) {
            weatherStation.setId(gatewayDeviceWeather.getDeviceId());
        } else {
            weatherStation.setId(device.getDeviceId());
        }
        weatherStation.setAirPressure(weather.getPressure());
        weatherStation.setHumidity(weather.getHumidity());
        weatherStation.setPm(weather.getAir_pm25());
        weatherStation.setRadiation("0");
        weatherStation.setRainfallAmount(weather.getWea());
        weatherStation.setTemperature(weather.getTem());
        weatherStation.setWindDirection(weather.getWin());
        weatherStation.setWindPower(weather.getWin_meter());
        weatherStation.setWindSpeed(weather.getWin_speed());

        grain.setWeatherStation(JSONObject.toJSONString(weatherStation));

        //封装好的数据
        log.info("---粮情信息封装完成-开始执行推送");

        reqData.setData(JSONObject.toJSONString(grain));

        doPushGrain(reqData);
    }


    private void doPushGrain(BaseReqData reqData) {

        GatewayDeviceReportService reportService = gatewayRemoteManager.getDeviceReportService(reqData.getDevice().getPushProtocol());
        if (null == reportService) {
            log.error("------------粮情推送失败，系统不存在当前协议执行类----{}", reqData.getDevice().getDeviceName());
            return;
        }

        reportService.reportGrainData(reqData);
    }


    /**
     * è®¡ç®—粮情温度 3组温度
     *
     * @param tempHex
     * @return
     */
    public static double getGrainTemp(String tempHex) {
        String strAscii = "";
        strAscii += toAscii(tempHex.substring(0, 2));
        strAscii += toAscii(tempHex.substring(2, 4));
        strAscii += toAscii(tempHex.substring(4, 6));

        Integer value = Integer.valueOf(strAscii);

        return value / 10.0;
    }

    private void add2Map(String deviceSn, THDto thDto) {
        contextMapTH.put("TH_" + deviceSn, thDto);
    }

    private THDto getTH(String deviceSn) {
        return contextMapTH.get("TH_" + deviceSn);
    }


    public static void main(String[] args) {
        String strMsg = "31 30 35 31 37 32 32 31 34 32 30 33 31 30 38 31 36 34 32 32 35 32 31 38 31 30 34 31 36 36 32 32 33 32 31 34 31 30 32 31 35 39 32 31 37 32 30 34 31 30 37 31 33 37 32 32 34 32 31 37 31 31 30";
        strMsg = strMsg.replaceAll(" ", "");
        System.out.println("---" + strMsg);
        int num = 20;
        String tempStr = "";
        double tem;
        for (int i = 0; i < num; i++) {
            tempStr = strMsg.substring(i * 3 * 2, i * 3 * 2 + 3 * 2);
            System.out.println("---" + tempStr);
            tem = getGrainTemp(tempStr);
            System.out.println("-tem--" + tem);
        }
    }

    public static char toAscii(String hexString) {
        int decimal = Integer.parseInt(hexString, 16);
        return (char) decimal;
    }

}

 src/main/java/com/fzzy/protocol/youxian1/client/ClientEngine.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,114 @@
package com.fzzy.protocol.youxian1.client;

import com.ld.io.api.InvokeResult;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.oio.OioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.oio.OioSocketChannel;
import io.netty.handler.codec.bytes.ByteArrayDecoder;
import io.netty.handler.codec.bytes.ByteArrayEncoder;
import lombok.extern.slf4j.Slf4j;

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

/**
 * æ¸¸ä»™åˆ†åº“，通讯协议
 */
@Slf4j
public class ClientEngine {

    /**
     * å®¢æˆ·ç«¯æ¨¡å¼åˆ›å»ºçš„连接通道
     */
    public static Map<String, Channel> clientChannelMap = new HashMap<>();


    public static Channel defaultChannel;

    public static void start(String ip, Integer port) throws Exception {
        if (defaultChannel != null) {
            log.info("-----IP={},连接存在，直接使用", ip);
            return;
        }
        EventLoopGroup group = new OioEventLoopGroup();
        Bootstrap b = new Bootstrap();
        //默认长连接
        b.option(ChannelOption.SO_KEEPALIVE, true);
        b.group(group).channel(OioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch)
                            throws Exception {
                        ChannelPipeline p = ch.pipeline();
                        // å­—符串解码 å’Œ ç¼–码
                        p.addLast("decoder", new ByteArrayDecoder());
                        p.addLast("encoder", new ByteArrayEncoder());
                        // è‡ªå·±çš„逻辑Handler
                        p.addLast("handler", new ClientHandler());
                    }
                });

        // å‘起异步连接请求，绑定连接端口和host信息
        ChannelFuture channelFuture = b.connect(ip, port);

        defaultChannel = channelFuture.channel();

        // channelFuture.channel().closeFuture().sync();

        add2ChannelMap(ip, defaultChannel);

        channelFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture arg0) throws Exception {
                if (channelFuture.isSuccess()) {
                    log.info("-----IP={},连接成功", ip);
                } else {
                    log.info("-----IP={},连接失败，自动关闭线程", ip);
                    channelFuture.cause().printStackTrace();
                    group.shutdownGracefully(); // å…³é—­çº¿ç¨‹ç»„
                }
            }
        });
    }

    public static InvokeResult send(byte[] array) throws InterruptedException {

        if (null == defaultChannel) {
            return InvokeResult.SOCKET_NOT_CREATE;
        }
        if (!defaultChannel.isActive()) {
            return InvokeResult.CHANNEL_CLOSED;
        }
        defaultChannel.writeAndFlush(Unpooled.copiedBuffer(array)).sync();
        return InvokeResult.SUCCESS;
    }


    public static Channel getChannel() {
        return defaultChannel;
    }

    public static void add2ChannelMap(String key, Channel channel) {
        clientChannelMap.put(key, channel);
    }

    /**
     * å¦‚æžœ
     *
     * @param key
     * @return
     */
    public static Channel getChannel(String key) {
        Channel channel = clientChannelMap.get(key);
        if (null == channel) return null;
        if (channel.isActive()) {
            return channel;
        } else {
            channel.close();
        }
        return null;
    }
}

 src/main/java/com/fzzy/protocol/youxian1/client/ClientHandler.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,65 @@
package com.fzzy.protocol.youxian1.client;

import com.fzzy.api.utils.BytesUtil;
import com.fzzy.api.utils.SpringUtil;
import com.fzzy.protocol.youxian1.analysis.AnalysisService;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;


@Slf4j
public class ClientHandler extends SimpleChannelInboundHandler<Object> {

    private AnalysisService analysisService;

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        InetSocketAddress socketAddress = (InetSocketAddress) ctx.channel()
                .remoteAddress();

        log.info("连接终成功，IP={},port={}", socketAddress.getAddress()
                .getHostAddress(), socketAddress.getPort());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);

        InetSocketAddress socketAddress = (InetSocketAddress) ctx.channel().remoteAddress();

        log.info("连接终端掉线，IP={},port={}", socketAddress.getAddress(), socketAddress.getPort());


    }

    @Override
    public void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        byte[] bytes = (byte[]) msg;

        InetSocketAddress socketAddress = (InetSocketAddress) ctx.channel().remoteAddress();

        String strMsg = BytesUtil.bytesToString(bytes);

        log.info("终端返回信息，IP={},port={}，msg={}", socketAddress.getAddress(), socketAddress.getPort(), strMsg);

        if (null == analysisService) {
            analysisService = SpringUtil.getBean(AnalysisService.class);
        }

       try{
           analysisService.analysis(socketAddress.getAddress(), socketAddress.getPort(), strMsg);
       }catch (Exception e){
           log.error(e.getMessage(),e);
       }

    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        ctx.close();
    }

}

 src/main/java/com/fzzy/protocol/youxian1/package-info.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,4 @@
/**
 * æ¸¸ä»™åˆ†åº“
 */
package com.fzzy.protocol.youxian1;

 src/main/java/com/fzzy/protocol/youxian1/service/Youxian1GatewayGrainService.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,150 @@
package com.fzzy.protocol.youxian1.service;

import com.fzzy.api.data.GatewayDeviceProtocol;
import com.fzzy.api.utils.BytesUtil;
import com.fzzy.gateway.api.GatewaySyncGranService;
import com.fzzy.gateway.data.BaseReqData;
import com.fzzy.gateway.data.BaseResp;
import com.fzzy.gateway.entity.GatewayDevice;
import com.fzzy.protocol.ProtocolUtils;
import com.fzzy.protocol.youxian1.client.ClientEngine;
import com.ld.io.api.InvokeResult;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

/**
 * æ¸¸ä»™åˆ†åº“协议-粮情采集入口
 */
@Slf4j
@Component
public class Youxian1GatewayGrainService implements GatewaySyncGranService {


    @Override
    public String getGrainProtocol() {
        return GatewayDeviceProtocol.GRAIN_YOUXIAN1_2023.getCode();
    }

    /**
     * @param reqData
     * @return
     */
    @Override
    public synchronized BaseResp syncGrain(BaseReqData reqData) {

        BaseResp resp = new BaseResp();

        GatewayDevice device = reqData.getDevice();

        if (null == device) {
            resp.setCode(500);
            resp.setMsg("系统未获取到下行连接设备信息，无法执行");
            log.error("----------------系统未获取到下行连接设备信息,无法执行---------");
            return resp;
        }
        try {
            //Step è¯·æ±‚信息放入内存
            ProtocolUtils.addSyncReq2Map(device.getDepotIdSys(), reqData);

            // åˆ¤æ–­æ•°æ®æœ‰æ²¡æœ‰æ”¶å–完整
            String[] attCable = device.getCableRule().split("-");
            int cableZ = Integer.valueOf(attCable[0]);
            int cableY = Integer.valueOf(attCable[1]);
            int cableX = Integer.valueOf(attCable[2]);

            // ç”Ÿæˆç²®æƒ…信息
            String hexCmd = buildGrainCmd(device, cableX, cableY, cableZ);

            Channel channel = ClientEngine.getChannel(device.getIp());

            if (null == channel) {
                ClientEngine.start(device.getIp(), device.getPort());
                Thread.sleep(500);
            }

            InvokeResult message = ClientEngine.send(BytesUtil.hexStrToBytes(hexCmd));

            log.error("平台------>>>>主控：发送粮情检测命令-{}---{}", message, hexCmd);
            // å°è£…返回信息
            if (!InvokeResult.SUCCESS.getCode().equals(message.getCode())) {
                log.error("平台------>>>>控制柜：发送粮情检测命令-失败{}", message.getMessage());
                resp.setCode(500);
                resp.setMsg("平台------>>>>控制柜：发送粮情检测命令-失败：" + message.getMessage());
            }

        } catch (Exception e) {
            log.error("粮情检测异常：{}", e);
            resp.setCode(500);
            resp.setMsg("平台------>>>>控制柜：发送粮情检测命令：" + e.getMessage());
            return resp;
        }
        return resp;
    }


    /**
     * ï¼ˆå‘采集命令）： A5 A5 A5 80 8X 41 8Y 8Z 8M
     * 8 X--分机号(90--16号分机)  8Y---起始间  8Z---本仓间数  8M---每间点数
     * <p>
     * X--控制器号；Y--起始间；Z--结束间；M--点数
     * ç”Ÿæˆç²®æƒ…采集命令
     *
     * @param device
     * @return
     */
    private String buildGrainCmd(GatewayDevice device, int cableX, int cableY, int cableZ) {

        String hexCmd = "A5A5A580{id}41{start}{end}{num}";

        //分机号，待根据配置进行调整，暂时未知计算规则 TODO
        String hexId = "90";
        hexCmd = hexCmd.replace("{id}", hexId);


        //开始列，默认从1开始
        String hexStart = "81";
        hexCmd = hexCmd.replace("{start}", hexStart);

        //截至列
        String hexEnd = "8" + cableX;
        hexCmd = hexCmd.replace("{end}", hexEnd);

        //点位数，目前未知具体计算规则 TODO
        int num = cableX * cableY * cableZ;
        String hexNum = BytesUtil.intToHexStr1(num);
        hexCmd = hexCmd.replace("{num}", hexNum);

        return hexCmd;
    }

    @Override
    public BaseResp syncGrainTh(BaseReqData reqData) {
        return new BaseResp();
    }

    @Override
    public BaseResp syncConf(BaseReqData reqData) {
        return new BaseResp();
    }

    @Override
    public BaseResp writeConf(BaseReqData reqData) {
        return new BaseResp();
    }

    @Override
    public BaseResp initCable(BaseReqData reqData) {
        return new BaseResp();
    }

    @Override
    public BaseResp disconnect(BaseReqData reqData) {
        return new BaseResp();
    }

    @Override
    public BaseResp transparent(BaseReqData reqData) {
        return new BaseResp();
    }
}