1、去掉基站GGA的处理

sec-beidou-rtcm/src/main/java/com/imdroid/sideslope/bd/Gga.java

@@ -155,6 +155,7 @@ public class Gga {
     public boolean isFixed(){
         //Possible values for quality: 0 = No fix, 1 = autonomous GNSS fix, 2 = differential GNSS fix, 4 = RTK fixed, 5 = RTK float, 6 =
         //estimated/dead reckoning fix
-        return (quality==1 || quality==2 || quality==4);
+        //return (quality==1 || quality==2 || quality==4);
+        return (quality==4);
     }
 }

sec-beidou-rtcm/src/main/java/com/imdroid/sideslope/executor/D331RtcmMessageExecutor.java

@@ -1,12 +1,10 @@
 package com.imdroid.sideslope.executor;
 
-import com.imdroid.common.util.ThreadManager;
 import com.imdroid.secapi.client.BeidouClient;
 import com.imdroid.secapi.dto.GnssDevice;
 import com.imdroid.sideslope.bd.Gga;
 import com.imdroid.sideslope.calc.GNSSDataCalcService;
 import com.imdroid.sideslope.message.D331RtcmMessage;
-import com.imdroid.sideslope.message.D341LocationMessage;
 import com.imdroid.sideslope.sal.Device;
 import com.imdroid.sideslope.sal.DeviceService;
 import com.imdroid.sideslope.server.DeviceChannel;
@@ -34,8 +32,6 @@ public class D331RtcmMessageExecutor implements Executor<D331RtcmMessage, Void>
     private DeviceService deviceService;
     @Autowired
     private BeidouClient beidouClient;
-    @Autowired
-    private GNSSDataCalcService gnssCalcService;
 
     @Override
     public Void execute(D331RtcmMessage message) {
@@ -79,7 +75,7 @@ public class D331RtcmMessageExecutor implements Executor<D331RtcmMessage, Void>
                 device1.setAltitude(gga.getAltitude());
             }
             // 借D341做GGA的统计分析
-            D341LocationMessage d341Message = new D341LocationMessage();
+            /*D341LocationMessage d341Message = new D341LocationMessage();
             d341Message.setGga(gga);
             d341Message.setId(message.getId());
             d341Message.setTenantId(message.getTenantId());
@@ -88,7 +84,7 @@ public class D331RtcmMessageExecutor implements Executor<D331RtcmMessage, Void>
             d341Message.setB562_loc(null);
             ThreadManager.getFixedThreadPool().submit(() -> {
                 gnssCalcService.calcSingle(d341Message,true);
-            });
+            });*/
         }
 
         // 收到第一个数据包，如果控制通道没连接，也通知上线

sec-test-device/src/main/java/com/imdroid/beidou/test_device/BeidouTestApp.java

@@ -1,6 +1,5 @@
 package com.imdroid.beidou.test_device;
 
-import com.imdroid.beidou.test_device.task.BeidouDevice;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.context.annotation.ComponentScan;
@@ -17,8 +16,8 @@ public class BeidouTestApp {
     public static void main(String[] args) {
 
         SpringApplication.run(BeidouTestApp.class, args);
-        BeidouDevice beidouDevice = new BeidouDevice();
+        /*BeidouDevice beidouDevice = new BeidouDevice();
         beidouDevice.connectServer();
-        beidouDevice.run();
+        beidouDevice.run();*/
     }
 }

sec-test-device/src/main/java/com/imdroid/beidou/test_device/service/OnlineChannels.java

@@ -0,0 +1,39 @@
+package com.imdroid.beidou.test_device.service;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * @author Layton
+ * @date 2023/2/2 21:00
+ * 提供两种获取channel的方法：通过deviceId；通过imei。imei是DTU连接服务器最先上报的消息
+ * 利用imei-ipaddr-deviceId，可以发现imei和deviceId的绑定关系
+ */
+public class OnlineChannels {
+
+    public static final OnlineChannels INSTANCE = new OnlineChannels();
+
+    // 设备已连接，deviceId已上报
+    private final Map<String, Channel> dataChannels = new ConcurrentHashMap<>();
+
+    private OnlineChannels() {}
+
+    public void updateChannel(String deviceId, Channel channel) {
+        dataChannels.put(deviceId, channel);
+    }
+
+    public void writeAndFlush(byte[] data){
+        for(Channel channel:dataChannels.values()){
+            if(channel.isActive()) {
+                ByteBuf buf = Unpooled.buffer();
+                buf.writeBytes(data);
+                channel.writeAndFlush(buf);
+            }
+        }
+    }
+
+}

sec-test-device/src/main/java/com/imdroid/beidou/test_device/service/tcp/RtcmTcpHandler.java

@@ -0,0 +1,62 @@
+package com.imdroid.beidou.test_device.service.tcp;
+
+import com.imdroid.beidou.test_device.service.OnlineChannels;
+import com.imdroid.common.util.DataTypeUtil;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+@ChannelHandler.Sharable
+@Component
+public class RtcmTcpHandler extends SimpleChannelInboundHandler<ByteBuf> {
+    private final Logger logger = LoggerFactory.getLogger(RtcmTcpServer.class);
+    @Override
+    public void channelActive(ChannelHandlerContext ctx){
+        OnlineChannels.INSTANCE.updateChannel(ctx.channel().remoteAddress().toString(), ctx.channel());
+    }
+    @Override
+    public void channelRead0(ChannelHandlerContext ctx, ByteBuf src) throws Exception{
+        if (logger.isDebugEnabled()) {
+            byte[] data = new byte[src.readableBytes()];
+            src.getBytes(0, data);
+            logger.debug("receive message:" + DataTypeUtil.getHexString(data));
+        }
+        try {
+            OnlineChannels.INSTANCE.updateChannel(ctx.channel().remoteAddress().toString(), ctx.channel());
+        }
+        catch (Exception e){
+
+        }
+    }
+/*
+    @Override
+    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
+        if(evt instanceof IdleStateEvent) {
+            IdleStateEvent event = (IdleStateEvent) evt;
+
+            if (event.state() == IdleState.READER_IDLE) {
+                Channel channel = ctx.channel();
+                logger.info(channel.remoteAddress() + " idle too long to be closed");
+                channel.close();
+            }
+        }
+    }
+*/
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx)
+            throws Exception {
+        logger.info("channel inactive");
+        ctx.close();
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+        cause.printStackTrace();
+        ctx.close();
+    }
+
+}

sec-test-device/src/main/java/com/imdroid/beidou/test_device/service/tcp/RtcmTcpServer.java

@@ -0,0 +1,59 @@
+package com.imdroid.beidou.test_device.service.tcp;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.handler.logging.LogLevel;
+import io.netty.handler.logging.LoggingHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.stereotype.Component;
+
+@Component
+public class RtcmTcpServer implements ApplicationRunner {
+    private final Logger logger = LoggerFactory.getLogger(RtcmTcpServer.class);
+
+    @Value("${netty.test.port}")
+    private int port;
+
+    @Override
+    public void run(ApplicationArguments args) throws Exception {
+        new Thread(this::start0, "test-tcp-server").start();
+    }
+
+    private void start0() {
+        logger.info("test tcp server starting...");
+        // Configure the server.
+        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
+        EventLoopGroup workerGroup = new NioEventLoopGroup();
+        try {
+            ServerBootstrap b = new ServerBootstrap();
+            b.group(bossGroup, workerGroup)
+                    .channel(NioServerSocketChannel.class)
+                    .handler(new LoggingHandler(LogLevel.INFO))
+                    .childHandler(new ChannelInitializer() {
+                        @Override
+                        protected void initChannel(Channel channel) throws Exception {
+                            ChannelPipeline p = channel.pipeline();
+                            //p.addLast(new IdleStateHandler(600, 600, 600));    //设置心跳超时时间，秒
+                            p.addLast(new RtcmTcpHandler());
+                        }
+                    });
+            Channel ch = b.bind(port).sync().channel();
+            logger.info("tcp server start at port {}", port);
+            ch.closeFuture().sync();
+        } catch (Exception el) {
+            logger.error(el.toString());
+        } finally {
+            bossGroup.shutdownGracefully();
+            workerGroup.shutdownGracefully();
+        }
+    }
+}

sec-test-device/src/main/java/com/imdroid/beidou/test_device/service/tcp/TCPClient.java

@@ -1,4 +1,4 @@
-package com.imdroid.beidou.test_device.service;
+package com.imdroid.beidou.test_device.service.tcp;
 
 import com.imdroid.common.util.ThreadManager;
 import io.netty.bootstrap.Bootstrap;

sec-test-device/src/main/java/com/imdroid/beidou/test_device/service/tcp/TCPListener.java

@@ -1,4 +1,4 @@
-package com.imdroid.beidou.test_device.service;
+package com.imdroid.beidou.test_device.service.tcp;
 
 public interface TCPListener {
     void onConnected();

sec-test-device/src/main/java/com/imdroid/beidou/test_device/service/tcp/TcpMessageHandler.java

@@ -1,4 +1,4 @@
-package com.imdroid.beidou.test_device.service;
+package com.imdroid.beidou.test_device.service.tcp;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;

sec-test-device/src/main/java/com/imdroid/beidou/test_device/service/udp/UDPClient.java

@@ -1,4 +1,4 @@
-package com.imdroid.beidou.test_device.service;
+package com.imdroid.beidou.test_device.service.udp;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

sec-test-device/src/main/java/com/imdroid/beidou/test_device/task/BeidouDevice.java

@@ -1,6 +1,6 @@
 package com.imdroid.beidou.test_device.task;
 
-import com.imdroid.beidou.test_device.service.UDPClient;
+import com.imdroid.beidou.test_device.service.udp.UDPClient;
 import com.imdroid.common.util.ByteUtil;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;

sec-test-device/src/main/java/com/imdroid/beidou/test_device/task/BeidouTCPDevice.java

@@ -0,0 +1,79 @@
+package com.imdroid.beidou.test_device.task;
+
+import com.imdroid.beidou.test_device.service.OnlineChannels;
+import com.imdroid.common.util.ByteUtil;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.annotation.EnableScheduling;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+import java.io.BufferedReader;
+import java.io.FileReader;
+import java.io.IOException;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+
+@Component
+@Configuration
+@EnableScheduling
+public class BeidouTCPDevice {
+    FileReader fr=null;
+    BufferedReader br=null;
+    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
+    int cycleCount = 0;
+
+    @Scheduled(cron = "0 6,16,26,36,46,56 * * * ?") // 每10分钟执行一次
+    public void execute() throws IOException {
+        if(br == null) {
+            try{
+                String dataFileName = "C:\\Users\\wd\\Desktop\\log\\2307046_0562_6xx.log";
+                fr = new FileReader(dataFileName);
+                br = new BufferedReader(fr);
+            }
+            catch (Exception e){
+                return;
+            }
+        }
+
+        try {
+            String line = "";
+            String[] arrs = null;
+
+            long lastTime = 0;
+            int d331Count = 0;
+
+            while ((line = br.readLine()) != null) {
+                arrs = line.split(" ");
+                if(arrs.length != 3) continue;
+                Date time = sdf.parse(arrs[0] + " " + arrs[1]);
+
+                if (lastTime!=0 && Math.abs(time.getTime() - lastTime) > 60 * 1000) {//超过1分钟为一个周期
+                    System.out.println(time+" cycle "+cycleCount+", b562 num "+d331Count);
+                    cycleCount++;
+                    break;
+                }
+
+                if(arrs[2].startsWith("d331")) {
+                    OnlineChannels.INSTANCE.writeAndFlush(ByteUtil.hexStringTobyte(arrs[2]));
+                    d331Count++;
+                    Thread.sleep(980);
+                }
+                lastTime = time.getTime();
+
+            }
+            if(line == null) {
+                br.close();
+                fr.close();
+                br = null;
+                fr = null;
+                System.out.println("finish!");
+            }
+        }
+        catch (Exception e){
+            e.printStackTrace();
+            if(br!=null) br.close();
+            if(fr!=null) fr.close();
+        }
+    }
+
+}