吴家粮库协议解析-1

jiazx0107@163.com

2023-12-29 dfe62588ca792d4a5eb3e2722675e7263aa46a4e

吴家粮库协议解析-1

 src/main/java/com/fzzy/api/data/GatewayDeviceProtocol.java

@@ -17,6 +17,7 @@
    GRAIN_FZZY_ZLDZ_WEB("GRAIN_FZZY_ZLDZ_WEB", "粮情-FZZY-ZLDZ网口协议"),
    GRAIN_YOUXIAN0_2023("GRAIN_YOUXIAN0_2023", "粮情-游仙主库协议"),
    GRAIN_YOUXIAN1_2023("GRAIN_YOUXIAN1_2023", "粮情-游仙分库协议"),
    GRAIN_WUJIA_2023("GRAIN_WUJIA_2023", "粮情-吴家粮库协议"),
    DEVICE_WEIGHT_HTTP("DEVICE_WEIGHT_HTTP", "地磅-HTTP协议"),
    DEVICE_WEIGHT_TCP_YH("DEVICE_WEIGHT_TCP_YH", "地磅-耀华TCP协议"),
    DEVICE_IDCARD_HTTP("DEVICE_IDCARD_HTTP", "身份证-HTTP协议"),

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

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,232 @@
package com.fzzy.protocol.wujia.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 lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.List;

@Slf4j
@Component(AnalysisService.BEAN_ID)
public class AnalysisService {

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


    @Resource
    private GatewayRemoteManager gatewayRemoteManager;


    /**
     * ç²®æƒ…接线
     *
     * @param address
     * @param port
     * @param strMsg
     */
    public void analysis(InetAddress address, int port, String strMsg) {

        if (strMsg.startsWith("FE")) {

            //DO NOTHING
        } else {
            this.analysisGrainStep1(address, strMsg);
        }
    }

    /**
     * ç²®æƒ…解析
     * X1 data1L data1H data2L data2H data3L data3H………datanL datanH DATA1H DATA1L DATA2H DATA2L AA BB CC
     * X1---发送的序号  AA BB CC----结束标志
     * DATA1H DATA1L----仓内湿度
     * DATA2H DATA2L----仓内温度
     * data1L data1H-----18B20返回温度
     * æ³¨æ„ï¼šç²®æ¸©è¿”回数据高位在后，低位在前，转换为十进制后乘以0.0625，得到实际温度值
     * ä»“温湿度高位在前，低位在后，转换为十进制后乘以0.1得到实际温湿度值
     *
     * @param strMsg
     */
    private void analysisGrainStep1(InetAddress address, String strMsg) {
        int start = 0;
        int len = 1 * 2;
        String tag = strMsg.substring(start, start + 2);
        int depotIdSys = BytesUtil.hexToInt(tag);

        BaseReqData reqData = ProtocolUtils.getSyncReq(depotIdSys + "");

        if (null == reqData) {
            log.error("---------没有获取到请求信息，不执行解析------{}", address);
            return;
        }

        //仓内湿度
        start = 1 * 2;
        len = 2 * 2;
        tag = strMsg.substring(start, start + len);
        double hIn = BytesUtil.hexToInt(tag) * 0.1;

        //仓内温度
        start = 3 * 2;
        len = 2 * 2;
        tag = strMsg.substring(start, start + len);
        double tIn = BytesUtil.hexToInt(tag) * 0.1;

        GatewayDevice device = reqData.getDevice();

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

        int sumNum = cableZ * cableY * cableX;
        //粮温信息
        start = 5 * 2;
        len = 2 * 2;
        List<Double> points = new ArrayList<>();
        double temp = 0.0;
        for (int i = 0; i < sumNum; i++) {
            start = start + i * 2 * 2;
            tag = strMsg.substring(start, start + len);
            temp = BytesUtil.hexToInt(BytesUtil.tran_LH(tag)) * 0.0625;
            temp = NumberUtil.keepPrecision(temp, 1);
            points.add(temp);
        }

        //执行封装解析
        analysisGrainStep2(reqData, cableZ, cableY, cableX, points, hIn, tIn);
    }

    private void analysisGrainStep2(BaseReqData reqData, int cableZ, int cableY, int cableX, List<Double> points, double hIn, double tIn) {
        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<>();


        ths.add(new GrainTH(tIn + "", hIn + "", "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);
    }
}

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

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

import com.fzzy.api.utils.BytesUtil;
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 implements Runnable {

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


    private String host;
    private int port;
    private Channel defaultChannel;


    public ClientEngine(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public void start() {
        Thread thread = new Thread(this);
        thread.start();
    }

    @Override
    public void run() {
        try {
            startClient();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void startClient() throws Exception {

        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(host, port);
        this.defaultChannel = channelFuture.channel();
        // channelFuture.channel().closeFuture().sync();

        this.add2ChannelMap(host, defaultChannel);

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

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

    public static InvokeResult send2(String hex, Channel channel) throws InterruptedException {
        if (null == channel) {
            return InvokeResult.SOCKET_NOT_CREATE;
        }
        if (!channel.isActive()) {
            return InvokeResult.CHANNEL_CLOSED;
        }
        channel.writeAndFlush(Unpooled.copiedBuffer(BytesUtil.hexStrToBytes(hex))).sync();
        return InvokeResult.SUCCESS;
    }


}

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

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

import com.fzzy.api.utils.BytesUtil;
import com.fzzy.api.utils.SpringUtil;
import com.fzzy.protocol.wujia.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/wujia/package-info.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,4 @@
/**
 * å´å®¶ç²®åº“2023
 */
package com.fzzy.protocol.wujia;

 src/main/java/com/fzzy/protocol/wujia/service/WujiaGatewayGrainService.java

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

import com.fzzy.api.data.GatewayDeviceProtocol;
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.youxian0.client.ClientEngine;
import com.ld.io.api.InvokeResult;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

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

/**
 * ç²®æƒ…采集入口
 */
@Slf4j
@Component
public class WujiaGatewayGrainService implements GatewaySyncGranService {


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


    /**
     * ä»“库命令
     */
    public static Map<String, String> depotGrainCmdMap = new HashMap<>();


    static {
        depotGrainCmdMap.put("KEY_1", "FC0101010314030100BDDB");
        depotGrainCmdMap.put("KEY_2", "FC0102040318030200BDDB");
        depotGrainCmdMap.put("KEY_3", "FC0203010318040100BDDB");
        depotGrainCmdMap.put("KEY_4", "FC0304010318040100BDDB");
        depotGrainCmdMap.put("KEY_5", "FC0305040318040200BDDB");
        depotGrainCmdMap.put("KEY_6", "FC0406010210030100BDDB");
        depotGrainCmdMap.put("KEY_7", "FC0407030314030200BDDB");
        depotGrainCmdMap.put("KEY_8", "FC0408060314030300BDDB");
        depotGrainCmdMap.put("KEY_9", "FC050901042A040100BDDB");
        depotGrainCmdMap.put("KEY_10", "FC050A010324040100BDDB");
        depotGrainCmdMap.put("KEY_11", "FC060B04042A040200BDDB");
    }

    /**
     * 1仓： FC 01 01 01 03 14 03 01 00 BD DB
     * 2仓： FC 01 02 04 03 18 03 02 00 BD DB
     * 3仓： FC 02 03 01 03 18 04 01 00 BD DB
     * 4仓： FC 03 04 01 03 18 04 01 00 BD DB
     * 5仓： FC 03 05 04 03 18 04 02 00 BD DB
     * 6仓： FC 04 06 01 02 10 03 01 00 BD DB
     * 7仓： FC 04 07 03 03 14 03 02 00 BD DB
     * 8仓： FC 04 08 06 03 14 03 03 00 BD DB
     * 9仓： FC 05 09 01 04 2A 04 01 00 BD DB
     * 10仓：FC 05 0A 01 03 24 04 01 00 BD DB
     * 11仓：FC 06 0B 04 04 2A 04 02 00 BD DB
     *
     * @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 hexStr = buildGrainCmd(device);

            Channel channel = null;
            if (null == channel) {
                ClientEngine clientEngine = new ClientEngine(device.getIp(), device.getPort());
                clientEngine.start();
                Thread.sleep(300);
                channel = clientEngine.getChannel();
            }

            InvokeResult message = ClientEngine.send2(hexStr, channel);

            log.error("平台------>>>>主控：发送粮情检测命令-{}---{}", message, hexStr);

            // å°è£…返回信息
            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;
    }

    @Override
    public BaseResp syncGrainTh(BaseReqData reqData) {

        BaseResp resp = new BaseResp();
        return resp;
    }

    /**
     * ç”Ÿæˆç²®æƒ…采集命令
     *
     * @param device
     * @return
     */
    private String buildGrainCmd(GatewayDevice device) {

        String key = "KEY_" + device.getDepotIdSys();

        return depotGrainCmdMap.get(key);

    }

    @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();
    }
}