城市大脑对接-推送过车数据

wj-realtime-computing/src/main/java/com/wanji/indicators/entity/BaseLaneDetector.java

+package com.wanji.indicators.entity;
+
+import lombok.Data;
+
+@Data
+public class BaseLaneDetector {
+    private int id;
+    private String geom;
+    private Integer gid;
+    private String xqid;
+    private String flbm;
+    private String lkid;
+    private String ldid;
+    private String qhdid;
+    private String cdid;
+}

wj-realtime-computing/src/main/java/com/wanji/indicators/mapper/BaseLaneDetectorMapper.java

+package com.wanji.indicators.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.wanji.indicators.entity.BaseLaneDetector;
+
+import java.util.List;
+
+public interface BaseLaneDetectorMapper extends BaseMapper<BaseLaneDetector>{
+
+
+	List<BaseLaneDetector> selectSense();
+}
+

wj-realtime-computing/src/main/java/com/wanji/indicators/service/BaseLaneDetectorService.java

+package com.wanji.indicators.service;
+
+import com.baomidou.mybatisplus.extension.service.IService;
+import com.wanji.indicators.entity.BaseLaneDetector;
+
+import java.util.List;
+
+/**
+* <p>
+* 车道基础信息 服务类
+* </p>
+*
+* @author fengyi
+* @since 2023-04-06
+*/
+public interface BaseLaneDetectorService extends IService<BaseLaneDetector> {
+
+    List<BaseLaneDetector> getSenseLaneDetectors();
+
+}

wj-realtime-computing/src/main/java/com/wanji/indicators/service/impl/BaseLaneDetectorServiceImpl.java

+package com.wanji.indicators.service.impl;
+
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.wanji.indicators.entity.BaseLaneDetector;
+import com.wanji.indicators.mapper.BaseLaneDetectorMapper;
+import com.wanji.indicators.service.BaseLaneDetectorService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.util.List;
+
+/**
+ * @author Kent HAN
+ * @date 2024/3/27 10:24
+ */
+@Service
+@Slf4j
+public class BaseLaneDetectorServiceImpl extends ServiceImpl<BaseLaneDetectorMapper, BaseLaneDetector> implements BaseLaneDetectorService {
+
+    @Resource
+    private BaseLaneDetectorMapper baseLaneDetectorMapper;
+
+    @Override
+    public List<BaseLaneDetector> getSenseLaneDetectors() {
+        return baseLaneDetectorMapper.selectSense();
+    }
+}

wj-realtime-computing/src/main/java/com/wanji/indicators/source/LaneDetectorSource.java

+package com.wanji.indicators.source;
+
+import com.wanji.indicators.entity.BaseLaneDetector;
+import com.wanji.indicators.service.BaseLaneDetectorService;
+import com.wanji.indicators.service.impl.BaseLaneDetectorServiceImpl;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 车道信息查询
+ */
+@Service
+public class LaneDetectorSource extends RichParallelSourceFunction<List<BaseLaneDetector>> {
+
+    private static final Logger log = LoggerFactory.getLogger(LaneDetectorSource.class);
+
+    private volatile boolean running = true;
+
+    BaseLaneDetectorService baseLaneDetectorService;
+
+    public void open(Configuration parameters) throws Exception {
+        super.open(parameters);
+        ApplicationContext beanConf = new ClassPathXmlApplicationContext("spring-container.xml");
+        baseLaneDetectorService = beanConf.getBean(BaseLaneDetectorServiceImpl.class);
+    }
+
+
+    @Override
+    public void run(SourceContext<List<BaseLaneDetector>> sourceContext) throws Exception {
+        try {
+            while (running) {
+                List<BaseLaneDetector> baseLaneDetectorList = baseLaneDetectorService.getSenseLaneDetectors();
+                sourceContext.collect(baseLaneDetectorList);
+                TimeUnit.MINUTES.sleep(1);
+            }
+        } catch (Exception e) {
+            log.error("查询线圈信息异常: " + e.getMessage(), e);
+        }
+    }
+
+    @Override
+    public void cancel() {
+        running = false;
+    }
+}

wj-realtime-computing/src/main/java/com/wanji/indicators/task/citybrain/devicestatus/DeviceStatusMain.java

@@ -5,6 +5,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 import com.wanji.indicators.task.citybrain.util.HttpUtil;
 import com.wanji.indicators.task.citybrain.vo.DeviceStatusVO;
 import com.wanji.indicators.task.citybrain.vo.RequestBodyVO;
+import com.wanji.indicators.util.PropertiesHelper;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.functions.sink.SinkFunction;
@@ -24,6 +25,7 @@ import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
+import java.util.Properties;
 
 /**
  * @author Kent HAN
@@ -34,6 +36,15 @@ public class DeviceStatusMain {
     private static final Logger log = LoggerFactory.getLogger(DeviceStatusMain.class);
     private static final ObjectMapper mapper = new ObjectMapper();
 
+    public static final Properties properties = PropertiesHelper.getInstance().getProperties();
+
+    // 城市大脑 URL 路径
+    public static final String urlPath;
+
+    static {
+        urlPath = properties.getProperty("city.brain.device.status");
+    }
+
     public static void main(String[] args) throws Exception {
 
         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
@@ -57,7 +68,7 @@ public class DeviceStatusMain {
                         AsyncHttpClient asyncHttpClient = Dsl.asyncHttpClient();
                         try {
                             BoundRequestBuilder boundRequestBuilder =
-                                    HttpUtil.prepareHttpPost(asyncHttpClient);
+                                    HttpUtil.prepareHttpPost(asyncHttpClient, urlPath);
 
                             String jsonBody = buildJsonFromValue(value);
                             boundRequestBuilder
@@ -79,7 +90,7 @@ public class DeviceStatusMain {
                     }
 
                     private String buildJsonFromValue(List<Row> rows) throws JsonProcessingException {
-                        RequestBodyVO requestBodyVO = HttpUtil.createRequestBodyVO();
+                        RequestBodyVO requestBodyVO = HttpUtil.createRequestBodyVO(urlPath);
                         requestBodyVO.setBody(convertBodyStr(rows));
                         // 将RequestBodyVO对象转换为JSON字符串
                         String jsonString = mapper.writeValueAsString(requestBodyVO);

wj-realtime-computing/src/main/java/com/wanji/indicators/task/citybrain/passcar/PassCarFlatMap.java

+package com.wanji.indicators.task.citybrain.passcar;
+
+/**
+ * @author fengyi
+ * @date 2023/10/30
+ * @description
+ */
+
+import com.vividsolutions.jts.geom.Coordinate;
+import com.vividsolutions.jts.geom.Geometry;
+import com.vividsolutions.jts.geom.Point;
+import com.wanji.indicators.entity.BaseLaneDetector;
+import com.wanji.indicators.model.CarTrackModel;
+import com.wanji.indicators.model.FrameModel;
+import com.wanji.indicators.task.citybrain.vo.CarPassVO;
+import com.wanji.indicators.util.CommonUtil;
+import com.wanji.indicators.util.GeomsConvertUtil;
+import org.apache.flink.api.common.state.MapState;
+import org.apache.flink.api.common.state.MapStateDescriptor;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.co.RichCoFlatMapFunction;
+import org.apache.flink.util.Collector;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class PassCarFlatMap extends RichCoFlatMapFunction<FrameModel, List<BaseLaneDetector>, CarPassVO> {
+
+    private List<BaseLaneDetector> baseLaneDetectorList = new ArrayList<>();
+
+    // 线圈状态，key-车辆ID，value-线圈数据
+    private transient MapState<Integer, CarPassVO> state;
+
+    @Override
+    public void open(Configuration config) {
+        MapStateDescriptor<Integer, CarPassVO> descriptor = new MapStateDescriptor<>("car-pass-state", Integer.class, CarPassVO.class);
+        state = getRuntimeContext().getMapState(descriptor);
+    }
+
+    @Override
+    public void flatMap1(FrameModel frameModel, Collector<CarPassVO> collector) throws Exception {
+
+        List<CarTrackModel> participantList = frameModel.getTrackList();
+        for (CarTrackModel participant : participantList) {
+            Integer participantId = participant.getId();
+            Double longitude = participant.getLongitude();
+            Double latitude = participant.getLatitude();
+
+            // 判断是否在线圈中
+            for (BaseLaneDetector detector : baseLaneDetectorList) {
+                String geom = detector.getGeom();
+                Geometry geometry = GeomsConvertUtil.fromHexWkb(geom);
+                geometry = geometry.convexHull();
+                Coordinate[] coordinates = geometry.getCoordinates();
+                double length = GeomsConvertUtil.getDistance(coordinates[0].x, coordinates[0].y, coordinates[1].x, coordinates[1].y);
+                double difLength = 3 - length;
+                if (difLength > 0) {
+                    geometry = geometry.buffer(GeomsConvertUtil.meterToDu(difLength / 2));
+                }
+
+                Point point = GeomsConvertUtil.createPoint(longitude + "," + latitude);
+                boolean inDetector = point.intersects(geometry);
+
+                if (inDetector) {
+                    // 如果在线圈内，则更新缓存
+                    CarPassVO carPassVO = new CarPassVO();
+                    carPassVO.setData_time(frameModel.getGlobalTimeStamp());
+                    carPassVO.setDeviceNo(detector.getId());
+
+                    Double angle = participant.getCourseAngle();
+                    int angleDir = CommonUtil.getAngleDir(angle);
+                    // 转换东西方向
+                    angleDir = CommonUtil.convertEastWest(angleDir);
+                    carPassVO.setSectionNo(angleDir);
+
+                    CarTrackModel.RoadNet roadnet = participant.getRoadnet();
+                    String laneId = roadnet.getLaneId();
+                    char c = laneId.charAt(laneId.length() - 1);
+                    String s = String.valueOf(c);
+                    if (CommonUtil.canConvertToInt(s)) {
+                        carPassVO.setLaneNo(Integer.parseInt(s));
+                    }
+                    carPassVO.setCoilNo(detector.getId());
+                    carPassVO.setTargetId(participantId);
+
+                    // 更新缓存
+                    state.put(participantId, carPassVO);
+                } else {
+                    // 如果不在线圈内，但是在缓存中，则收集数据
+                    CarPassVO carPassVO = state.get(participantId);
+                    if (carPassVO != null) {
+                        collector.collect(carPassVO);
+                        // 清除缓存
+                        state.remove(participantId);
+                    }
+                }
+            }
+        }
+    }
+
+    @Override
+    public void flatMap2(List<BaseLaneDetector> baseLaneDetectorList, Collector<CarPassVO> collector) throws Exception {
+        this.baseLaneDetectorList = baseLaneDetectorList;
+    }
+
+}
\ No newline at end of file

wj-realtime-computing/src/main/java/com/wanji/indicators/task/citybrain/passcar/PassCarMain.java

+package com.wanji.indicators.task.citybrain.passcar;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.wanji.indicators.model.FrameModel;
+import com.wanji.indicators.source.LaneDetectorSource;
+import com.wanji.indicators.task.citybrain.trajectory.TrajectoryMain;
+import com.wanji.indicators.task.citybrain.util.HttpUtil;
+import com.wanji.indicators.task.citybrain.vo.CarPassVO;
+import com.wanji.indicators.task.citybrain.vo.RequestBodyVO;
+import com.wanji.indicators.task.track.service.func.FrameFlatMap;
+import com.wanji.indicators.util.PropertiesHelper;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.connector.kafka.source.KafkaSource;
+import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
+import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.asynchttpclient.AsyncHttpClient;
+import org.asynchttpclient.BoundRequestBuilder;
+import org.asynchttpclient.Dsl;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Properties;
+
+/**
+ * 过车数据推送城市大脑
+ *
+ * @author Kent HAN
+ * @date 2024/3/27 9:18
+ */
+public class PassCarMain {
+
+    private static final Logger log = LoggerFactory.getLogger(TrajectoryMain.class);
+
+    private static final ObjectMapper mapper = new ObjectMapper();
+
+    public static final Properties properties = PropertiesHelper.getInstance().getProperties();
+
+    private static final PropertiesHelper instance = PropertiesHelper.getInstance();
+
+    // 城市大脑 URL 路径
+    public static final String urlPath;
+
+    static {
+        urlPath = properties.getProperty("city.brain.passcar.info");
+    }
+
+    public static void main(String[] args) throws Exception {
+
+        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+        // 获取配置文件中的kafka消费topic
+        String sourceTopic = properties.getProperty("consumer.topic");
+        String kafkaServerAddress = properties.getProperty("bootstrap.servers");
+
+        KafkaSource<String> source = KafkaSource.<String>builder()
+                .setProperties(instance.getConsumerProperties())
+                .setProperty("auto.offset.commit", "true")
+                .setProperty("auto.commit.interval.ms", "1000")
+                .setProperty("commit.offsets.on.checkpoint", "true")
+                .setBootstrapServers(kafkaServerAddress)
+                .setTopics(sourceTopic)
+                .setGroupId(properties.getProperty("consumer.group.id") + "-city-brain-passcar")
+                .setStartingOffsets(OffsetsInitializer.latest())
+                .setDeserializer(KafkaRecordDeserializationSchema.valueOnly(StringDeserializer.class))
+                .build();
+
+        DataStream<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(),
+                "city-brain-passcar-data-source");
+
+        SingleOutputStreamOperator<FrameModel> frameModelStream = stream
+                .flatMap(new FrameFlatMap())
+                .setParallelism(1)
+                .name("轨迹帧数据-JsonToObject");
+
+        SingleOutputStreamOperator<CarPassVO> passcarStream = frameModelStream
+                .keyBy(FrameModel::getGlobalTimeStamp)
+                .connect(env.addSource(new LaneDetectorSource()).broadcast())
+                .flatMap(new PassCarFlatMap())
+                .setParallelism(1)
+                .name("计算虚拟线圈过车数据");
+
+        passcarStream.addSink(new SinkFunction<CarPassVO>() {
+            @Override
+            public void invoke(CarPassVO carPassVO, Context context) throws Exception {
+                // 向城市大脑发送数据
+                AsyncHttpClient asyncHttpClient = Dsl.asyncHttpClient();
+                try {
+                    BoundRequestBuilder boundRequestBuilder =
+                            HttpUtil.prepareHttpPost(asyncHttpClient, urlPath);
+
+                    String jsonBody = buildJsonFromValue(carPassVO);
+                    boundRequestBuilder
+                            .setBody(jsonBody)
+                            .execute()
+                            .toCompletableFuture()
+                            .thenAccept(response -> {
+                                log.info("发送轨迹数据收到响应: " + response.getResponseBody());
+                            });
+                } catch (Exception e) {
+                    e.printStackTrace();
+                } finally {
+                    try {
+                        asyncHttpClient.close();
+                    } catch (IOException e) {
+                        e.printStackTrace();
+                    }
+                }
+            }
+        });
+
+        // 启动 Flink 作业
+        env.execute("向城市大脑发送轨迹");
+    }
+
+    private static String buildJsonFromValue(CarPassVO vo) throws JsonProcessingException {
+        RequestBodyVO requestBodyVO = HttpUtil.createRequestBodyVO(urlPath);
+        requestBodyVO.setBody(convertBodyStr(vo));
+        // 将RequestBodyVO对象转换为JSON字符串
+        String jsonString = mapper.writeValueAsString(requestBodyVO);
+        // 创建一个只包含一个元素的列表
+        return "[" + jsonString + "]";
+    }
+
+    private static String convertBodyStr(CarPassVO vo) throws JsonProcessingException {
+        return mapper.writeValueAsString(vo);
+    }
+}

wj-realtime-computing/src/main/java/com/wanji/indicators/task/citybrain/trajectory/TargetInfoFlatMap.java

-package com.wanji.indicators.task.citybrain.trajectory;
-
-import com.wanji.indicators.model.CarTrackModel;
-import com.wanji.indicators.model.FrameModel;
-import com.wanji.indicators.task.citybrain.vo.TargetInfoVO;
-import com.wanji.indicators.util.CommonUtil;
-import com.wanji.indicators.util.PropertiesHelper;
-import org.apache.flink.api.common.functions.FlatMapFunction;
-import org.apache.flink.util.Collector;
-import org.springframework.util.CollectionUtils;
-
-import java.util.ArrayList;
-import java.util.Date;
-import java.util.List;
-import java.util.Properties;
-
-/**
- * @author : jenny
- * @version : 1.0
- * @createTime : 2023/10/25 14:48
- * @Description :
- */
-public class TargetInfoFlatMap implements FlatMapFunction<FrameModel, TargetInfoVO> {
-
-    public static final Properties properties = PropertiesHelper.getInstance().getProperties();
-    public static final String serverIp;
-    public static final String serverMac;
-
-    static {
-        serverIp = properties.getProperty("server.ip");
-        serverMac = properties.getProperty("server.mac");
-    }
-
-    @Override
-    public void flatMap(FrameModel frameModel, Collector<TargetInfoVO> collector) throws Exception {
-
-        List<CarTrackModel> participantList = frameModel.getTrackList();
-        if(CollectionUtils.isEmpty(participantList)){
-            return;
-        }
-
-        List<TargetInfoVO> targetInfoVOList = new ArrayList<>();
-        String orgCode = frameModel.getOrgCode();
-
-        for (CarTrackModel participant : participantList) {
-            TargetInfoVO targetInfoVO = new TargetInfoVO();
-            targetInfoVO.setIpAddress(serverIp);
-            targetInfoVO.setMacAddress(serverMac);
-
-            Double angle = participant.getCourseAngle();
-            int angleDir = CommonUtil.getAngleDir(angle);
-            // 转换东西方向
-            angleDir = CommonUtil.convertEastWest(angleDir);
-            targetInfoVO.setSectionNo(angleDir);
-
-            CarTrackModel.RoadNet roadnet = participant.getRoadnet();
-            String laneId = roadnet.getLaneId();
-            char c = laneId.charAt(laneId.length() - 1);
-            String s = String.valueOf(c);
-            if (canConvertToInt(s)) {
-                targetInfoVO.setLaneNo(Integer.parseInt(s));
-            }
-
-            targetInfoVO.setData_time(new Date().getTime());
-            targetInfoVO.setDeviceNo(orgCode);
-            targetInfoVO.setId(participant.getId());
-
-            targetInfoVOList.add(targetInfoVO);
-        }
-
-        targetInfoVOList.forEach(collector::collect);
-    }
-
-    public static boolean canConvertToInt(String str) {
-        try {
-            Integer.parseInt(str);
-            return true;
-        } catch (NumberFormatException e) {
-            return false;
-        }
-    }
-}

wj-realtime-computing/src/main/java/com/wanji/indicators/task/citybrain/trajectory/TrajectoryMain.java

@@ -2,11 +2,13 @@ package com.wanji.indicators.task.citybrain.trajectory;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.wanji.indicators.model.CarTrackModel;
 import com.wanji.indicators.model.FrameModel;
 import com.wanji.indicators.task.citybrain.util.HttpUtil;
 import com.wanji.indicators.task.citybrain.vo.RequestBodyVO;
 import com.wanji.indicators.task.citybrain.vo.TargetInfoVO;
 import com.wanji.indicators.task.track.service.func.FrameFlatMap;
+import com.wanji.indicators.util.CommonUtil;
 import com.wanji.indicators.util.PropertiesHelper;
 import org.apache.flink.api.common.eventtime.WatermarkStrategy;
 import org.apache.flink.connector.kafka.source.KafkaSource;
@@ -16,20 +18,17 @@ import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.functions.sink.SinkFunction;
-import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
-import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
-import org.apache.flink.streaming.api.windowing.time.Time;
-import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
-import org.apache.flink.util.Collector;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.asynchttpclient.AsyncHttpClient;
 import org.asynchttpclient.BoundRequestBuilder;
 import org.asynchttpclient.Dsl;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.util.CollectionUtils;
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Date;
 import java.util.List;
 import java.util.Properties;
 
@@ -43,25 +42,21 @@ public class TrajectoryMain {
 
     private static final ObjectMapper mapper = new ObjectMapper();
 
-
     public static final Properties properties = PropertiesHelper.getInstance().getProperties();
-    // 城市大脑域名
-    public static final String domain;
-    // 城市大脑 URL 路径
-    public static final String urlPath;
-    public static final String appKey;
-    public static final String secretKey;
-    // 数据来源公司名称
-    public static final String from;
+
     private static final PropertiesHelper instance = PropertiesHelper.getInstance();
 
+    public static final String serverIp;
+
+    public static final String serverMac;
+
+    // 城市大脑 URL 路径
+    public static final String urlPath;
 
     static {
-        domain = properties.getProperty("city.brain.domain");
-        urlPath = properties.getProperty("city.brain.device.status");
-        appKey = properties.getProperty("city.brain.app.key");
-        secretKey = properties.getProperty("city.brain.secret.key");
-        from = properties.getProperty("city.brain.from");
+        serverIp = properties.getProperty("server.ip");
+        serverMac = properties.getProperty("server.mac");
+        urlPath = properties.getProperty("city.brain.target.info");
     }
 
     public static void main(String[] args) throws Exception {
@@ -92,58 +87,76 @@ public class TrajectoryMain {
                 .setParallelism(1)
                 .name("轨迹帧数据-JsonToObject");
 
-        SingleOutputStreamOperator<TargetInfoVO> targetInfoStream = frameModelStream
-                .flatMap(new TargetInfoFlatMap())
-                .setParallelism(1)
-                .name("将轨迹帧转换为交通大脑所需数据流");
-
-        targetInfoStream
-                .windowAll(TumblingProcessingTimeWindows.of(Time.milliseconds(100))).
-                apply(new AllWindowFunction<TargetInfoVO, List<TargetInfoVO>, TimeWindow>() {
-                    @Override
-                    public void apply(TimeWindow window, Iterable<TargetInfoVO> vos, Collector<List<TargetInfoVO>> out) {
-                        List<TargetInfoVO> res = new ArrayList<>();
-                        for (TargetInfoVO vo : vos) {
-                            res.add(vo);
-                        }
-                        out.collect(res);
+        frameModelStream.addSink(new SinkFunction<FrameModel>() {
+            @Override
+            public void invoke(FrameModel frameModel, Context context) throws Exception {
+
+                List<CarTrackModel> participantList = frameModel.getTrackList();
+                if(CollectionUtils.isEmpty(participantList)){
+                    return;
+                }
+
+                List<TargetInfoVO> targetInfoVOList = new ArrayList<>();
+                String orgCode = frameModel.getOrgCode();
+
+                for (CarTrackModel participant : participantList) {
+                    TargetInfoVO targetInfoVO = new TargetInfoVO();
+                    targetInfoVO.setIpAddress(serverIp);
+                    targetInfoVO.setMacAddress(serverMac);
+
+                    Double angle = participant.getCourseAngle();
+                    int angleDir = CommonUtil.getAngleDir(angle);
+                    // 转换东西方向
+                    angleDir = CommonUtil.convertEastWest(angleDir);
+                    targetInfoVO.setSectionNo(angleDir);
+
+                    CarTrackModel.RoadNet roadnet = participant.getRoadnet();
+                    String laneId = roadnet.getLaneId();
+                    char c = laneId.charAt(laneId.length() - 1);
+                    String s = String.valueOf(c);
+                    if (CommonUtil.canConvertToInt(s)) {
+                        targetInfoVO.setLaneNo(Integer.parseInt(s));
                     }
-                }).addSink(new SinkFunction<List<TargetInfoVO>>() {
-                    @Override
-                    public void invoke(List<TargetInfoVO> vos, Context context) {
-                        // 将收集到的数据发送至城市大脑
-                        AsyncHttpClient asyncHttpClient = Dsl.asyncHttpClient();
-                        try {
-                            BoundRequestBuilder boundRequestBuilder =
-                                    HttpUtil.prepareHttpPost(asyncHttpClient);
-
-                            String jsonBody = buildJsonFromValue(vos);
-                            boundRequestBuilder
-                                    .setBody(jsonBody)
-                                    .execute()
-                                    .toCompletableFuture()
-                                    .thenAccept(response -> {
-                                        log.info("发送设备在线状态检测数据收到响应: " + response.getResponseBody());
-                                    });
-                        } catch (Exception e) {
-                            e.printStackTrace();
-                        } finally {
-                            try {
-                                asyncHttpClient.close();
-                            } catch (IOException e) {
-                                e.printStackTrace();
-                            }
-                        }
 
+                    targetInfoVO.setData_time(new Date().getTime());
+                    targetInfoVO.setDeviceNo(orgCode);
+                    targetInfoVO.setId(participant.getId());
+
+                    targetInfoVOList.add(targetInfoVO);
+                }
+
+                // 将收集到的数据发送至城市大脑
+                AsyncHttpClient asyncHttpClient = Dsl.asyncHttpClient();
+                try {
+                    BoundRequestBuilder boundRequestBuilder =
+                            HttpUtil.prepareHttpPost(asyncHttpClient, urlPath);
+
+                    String jsonBody = buildJsonFromValue(targetInfoVOList);
+                    boundRequestBuilder
+                            .setBody(jsonBody)
+                            .execute()
+                            .toCompletableFuture()
+                            .thenAccept(response -> {
+                                log.info("发送轨迹数据收到响应: " + response.getResponseBody());
+                            });
+                } catch (Exception e) {
+                    e.printStackTrace();
+                } finally {
+                    try {
+                        asyncHttpClient.close();
+                    } catch (IOException e) {
+                        e.printStackTrace();
                     }
-                });
+                }
+            }
+        });
 
-        // 启动Flink作业
-        env.execute("推送设备在线状态");
+        // 启动 Flink 作业
+        env.execute("向城市大脑发送轨迹");
     }
 
     private static String buildJsonFromValue(List<TargetInfoVO> vos) throws JsonProcessingException {
-        RequestBodyVO requestBodyVO = HttpUtil.createRequestBodyVO();
+        RequestBodyVO requestBodyVO = HttpUtil.createRequestBodyVO(urlPath);
         requestBodyVO.setBody(convertBodyStr(vos));
         // 将RequestBodyVO对象转换为JSON字符串
         String jsonString = mapper.writeValueAsString(requestBodyVO);

wj-realtime-computing/src/main/java/com/wanji/indicators/task/citybrain/util/HttpUtil.java

@@ -21,8 +21,6 @@ public class HttpUtil {
     public static final Properties properties = PropertiesHelper.getInstance().getProperties();
     // 城市大脑域名
     public static final String domain;
-    // 城市大脑 URL 路径
-    public static final String urlPath;
     public static final String appKey;
     public static final String secretKey;
     // 数据来源公司名称
@@ -31,13 +29,12 @@ public class HttpUtil {
 
     static {
         domain = properties.getProperty("city.brain.domain");
-        urlPath = properties.getProperty("city.brain.device.status");
         appKey = properties.getProperty("city.brain.app.key");
         secretKey = properties.getProperty("city.brain.secret.key");
         from = properties.getProperty("city.brain.from");
     }
 
-    public static BoundRequestBuilder prepareHttpPost(AsyncHttpClient asyncHttpClient) {
+    public static BoundRequestBuilder prepareHttpPost(AsyncHttpClient asyncHttpClient, String urlPath) {
         long timestamp = new Date().getTime();
         String signature = makeSignature(timestamp);
 
@@ -71,7 +68,7 @@ public class HttpUtil {
     /**
      * 创建并初始化一个 RequestBodyVO 对象。
      */
-    public static RequestBodyVO createRequestBodyVO() {
+    public static RequestBodyVO createRequestBodyVO(String urlPath) {
         RequestBodyVO requestBodyVO = new RequestBodyVO();
         RequestBodyVO.Header header = new RequestBodyVO.Header();
 

wj-realtime-computing/src/main/java/com/wanji/indicators/task/citybrain/vo/CarPassVO.java

+package com.wanji.indicators.task.citybrain.vo;
+
+import lombok.Data;
+
+@Data
+public class CarPassVO {
+
+    // 数据时间, 毫秒级时间戳
+    private Long data_time;
+
+    // 设备编号
+    private Integer deviceNo;
+
+    // 断面编号，南向北 1，东向西 2，北向南 3，西向东 4
+    private Integer sectionNo;
+
+    // 车道编号，与车辆正常行驶方向的垂直方向从左向右依次编号，最左侧靠近道路中间线的为1车道
+    private Integer laneNo;
+
+    // 线圈编号
+    private Integer coilNo;
+
+    // 线圈状态，第几位表示几号线圈状态，状态1-到达，0-离开
+    private String wLoopState;
+
+    // 线圈状态掩码，掩码位为1对应wLoopState状态位有效，为0表示无效
+    private String wStateMask;
+
+    // 目标ID
+    private Integer targetId;
+
+    // 车长(m)
+    private Double targetLen;
+
+    // 目标类型，1行人，2非机动车，3小型车，4中型车，5大型车
+    private Integer targetType;
+
+    // 目标驶入线圈时的速度(km/h)
+    private Double speedIn;
+
+    // 目标驶离线圈时的速度(km/h)
+    private Double speedOut;
+
+    // 压占时间/存在时间（ms）
+    private Long presenceTime;
+
+    // 目标进入线圈的时间，毫秒级时间戳
+    private Long driveInTime;
+
+    // 驶离时间, 毫秒级时间戳
+    private Long driveOutTime;
+
+    // 车辆类型（字典见附录）
+    private String vehicleType;
+
+    // 车身颜色（字典见附录）
+    private String vehicleColor;
+
+    // 车牌类型（字典见附录）
+    private String plateType;
+
+    // 车牌号，如“苏A xxxx”
+    private String plateNumber;
+}

wj-realtime-computing/src/main/java/com/wanji/indicators/util/CommonUtil.java

@@ -322,4 +322,13 @@ public class CommonUtil {
         }
         return distance;
     }
+
+    public static boolean canConvertToInt(String str) {
+        try {
+            Integer.parseInt(str);
+            return true;
+        } catch (NumberFormatException e) {
+            return false;
+        }
+    }
 }

wj-realtime-computing/src/main/java/com/wanji/indicators/util/GeomsConvertUtil.java

@@ -3,6 +3,7 @@ package com.wanji.indicators.util;
 import com.alibaba.fastjson.JSONArray;
 import com.vividsolutions.jts.geom.*;
 import com.vividsolutions.jts.io.ParseException;
+import com.vividsolutions.jts.io.WKBReader;
 import com.vividsolutions.jts.io.WKTReader;
 import com.vividsolutions.jts.linearref.LengthLocationMap;
 import com.vividsolutions.jts.linearref.LinearLocation;
@@ -17,7 +18,6 @@ import org.apache.commons.lang.StringUtils;
 
 import java.awt.geom.GeneralPath;
 import java.awt.geom.Point2D;
-import java.io.StringWriter;
 import java.math.BigDecimal;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -1136,6 +1136,20 @@ public class GeomsConvertUtil {
         return new Coordinate(bjd, bwd);
     }
 
+    public static  Geometry fromHexWkb(String wkbHex)  {
+
+        WKBReader wkbReader = new WKBReader(geometryFactory);
+        byte[] wkb = WKBReader.hexToBytes(wkbHex);
+        Geometry g2 = null;
+        try {
+            g2 = wkbReader.read(wkb);
+        } catch (ParseException e) {
+            e.printStackTrace();
+        }
+
+        return g2;
+    }
+
     public static void main(String[] args) throws Exception {
 
         String wkt = "121.424661,31.39226;121.424881,31.39151;121.427819,31.381745;121.427953,31.381296;121.428082,31.380837;121.429196,31.376526;121.429994,31.373874;121.430299,31.372705;121.430537,31.371323;121.430723,31.370427;121.430914,31.369555;121.431103,31.368691;121.43127,31.367942;121.432941,31.362217;121.433261,31.360896;121.433515,31.35981;121.433691,31.358866;121.43398,31.35595;121.434063,31.355082;121.434143,31.354235;121.434701,31.348262;121.434975,31.347014;121.435106,31.346518;121.435187,31.346232;121.435625,31.345065;121.435902,31.344285;121.436606,31.342377;121.437375,31.340518;121.437711,31.339703;121.438044,31.338884;121.439232,31.33604;121.441546,31.33113;121.443878,31.326024;121.445757,31.321816;121.446752,31.31966;121.447063,31.318936;121.447369,31.318237;121.447905,31.316858;121.448035,31.316267;121.448187,31.315546;121.448293,31.314782;121.448497,31.313164;121.448623,31.310281;121.448696,31.307901;121.44866,31.307303;121.448642,31.306604;121.448625,31.305913;121.448931,31.300276;121.448984,31.299198;121.449126,31.298117;121.449373,31.296527;121.450076,31.293463;121.450251,31.292556;121.450429,31.291631;121.450841,31.289895;121.451269,31.287919;121.451429,31.286805;121.451543,31.285801;121.451693,31.284324;121.451924,31.28113;121.452023,31.279895;121.452124,31.278661;121.45231,31.277347;121.452461,31.276862;121.452637,31.276333;121.453007,31.275443;121.453265,31.274885;121.454004,31.273619;121.455329,31.271675;121.45583,31.27095;121.457568,31.268372;121.457939,31.267754;121.458115,31.267362;121.458265,31.26697;121.458421,31.265833;121.458557,31.263704;121.45866,31.26213;121.458817,31.260169;121.45895,31.259749;121.459204,31.258891;121.45946,31.258028;121.459485,31.257836;121.459485,31.257535;121.459496,31.256501;121.459479,31.255648;121.459418,31.254958;121.459025,31.252636;121.458995,31.252406;121.459004,31.252209;121.459112,31.250582;121.459083,31.250372;121.459011,31.250177;121.458846,31.25002;121.458636,31.249863;121.457204,31.249166;121.456868,31.248956;121.456573,31.248734;121.456389,31.248532;121.456215,31.248302;121.456047,31.247995;121.455684,31.247255;121.455313,31.2465;121.455125,31.246055;121.455018,31.245642;121.454982,31.245213;121.454998,31.244865;121.455065,31.244482;121.455181,31.244115;121.455346,31.243768;121.455529,31.243495;121.455743,31.243257;121.456017,31.242991;121.456234,31.242817;121.456465,31.242656;121.456712,31.242519;121.456953,31.242422;121.457267,31.242317;121.457648,31.242207;121.458699,31.241883;121.459971,31.241503;121.460583,31.241306;121.461344,31.24104;121.467182,31.23874;121.468151,31.238373;121.469041,31.238051;121.470426,31.237687;121.471569,31.237297;121.471853,31.237137;121.472116,31.236953;121.472481,31.236665;121.472974,31.236165;121.474106,31.234365;121.475137,31.232781;121.476134,31.231231;121.477547,31.22947;121.478086,31.228782;121.478159,31.228679;121.478189,31.228622;121.478233,31.22853;121.478277,31.228423;121.478386,31.228174;121.47843,31.227951;121.478461,31.227758;121.478468,31.227702;121.478473,31.22761;121.478472,31.22719;121.478448,31.226138;121.478445,31.225997;121.478438,31.225904;121.478417,31.225777;121.478378,31.22561;121.478332,31.225509;121.478246,31.225362;121.478035,31.225136;121.47761,31.224797;121.477185,31.224532;121.476407,31.224124;121.475127,31.223526;121.474212,31.223128;121.473306,31.222745;121.472383,31.222356;121.466695,31.22004;121.461164,31.217746;121.459957,31.217231;121.458745,31.216717;121.452162,31.213851;121.451048,31.213382;121.44972,31.212813;121.449563,31.212682;121.449412,31.212538;121.449314,31.212327;121.449314,31.212087;121.449296,31.211233;121.449211,31.210127;121.448814,31.208214;121.448792,31.208145;121.448652,31.207754;121.448576,31.207569;121.448494,31.207395;121.448466,31.207353;121.447669,31.206248;121.446883,31.205164;121.446424,31.204528;121.445965,31.203891;121.444296,31.201692;121.443138,31.200113;121.442022,31.198663;121.440277,31.196306;121.439249,31.194984;121.439126,31.194794;121.439068,31.194684;121.439012,31.194564;121.438969,31.194472;121.438133,31.191698;121.437404,31.189259;121.437235,31.188728;121.437119,31.188241;121.436961,31.187457;121.436745,31.186035;121.436528,31.184209;121.436272,31.182001;121.436151,31.180969;121.435985,31.179869;121.435917,31.179221;121.435513,31.175943;121.435218,31.173487;121.435111,31.17245;121.435031,31.171169;121.434848,31.169209;121.434702,31.168228;121.434551,31.167266;121.434476,31.166688;121.43411,31.163133;121.434067,31.162858;121.434014,31.162583;121.433201,31.159207;121.432658,31.157705;121.432278,31.157017;121.431802,31.156247;121.430796,31.155095;121.430041,31.154579;121.429217,31.154016;121.428206,31.15342;121.424098,31.151061;121.420987,31.148834;121.417147,31.145353;121.416352,31.144585;121.415073,31.143098;121.414604,31.142675;121.414107,31.142217;121.413622,31.141771;121.409971,31.138218;121.405856,31.133814;121.403605,31.131565;121.40291,31.1309;121.402241,31.130274;121.401049,31.129156;121.400066,31.128214;121.398684,31.126855;121.396102,31.124412;121.395084,31.123459;121.3947,31.123076;121.394223,31.122539;121.393845,31.122059;121.393524,31.121616;121.39302,31.120899;121.392492,31.120155;121.39125,31.118396;121.390298,31.117133;121.38802,31.114151;121.387549,31.113551;121.38586,31.11171;121.385373,31.111152";

wj-realtime-computing/src/main/resources/mapper/BaseLaneDetectorMapper.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="com.wanji.indicators.mapper.BaseLaneDetectorMapper">
+
+    <resultMap id="BaseResultMap" type="com.wanji.indicators.entity.BaseLaneDetector">
+        <id column="id" property="id" jdbcType="INTEGER" />
+        <result column="geom" property="geom" jdbcType="LONGVARCHAR" />
+        <result column="gid" property="gid" jdbcType="INTEGER" />
+        <result column="xqid" property="xqid" jdbcType="VARCHAR" />
+        <result column="flbm" property="flbm" jdbcType="VARCHAR" />
+        <result column="lkid" property="lkid" jdbcType="VARCHAR" />
+        <result column="ldid" property="ldid" jdbcType="VARCHAR" />
+        <result column="qhdid" property="qhdid" jdbcType="VARCHAR" />
+        <result column="cdid" property="cdid" jdbcType="VARCHAR" />
+    </resultMap>
+
+    <!-- Insert statement -->
+    <insert id="insert" parameterType="com.wanji.indicators.entity.BaseLaneDetector">
+        INSERT INTO t_base_lane_detector (id, geom, gid, xqid, flbm, lkid, ldid, qhdid, cdid)
+        VALUES (#{id,jdbcType=INTEGER}, #{geom,jdbcType=LONGVARCHAR}, #{gid,jdbcType=INTEGER},
+                #{xqid,jdbcType=VARCHAR}, #{flbm,jdbcType=VARCHAR}, #{lkid,jdbcType=VARCHAR},
+                #{ldid,jdbcType=VARCHAR}, #{qhdid,jdbcType=VARCHAR}, #{cdid,jdbcType=VARCHAR})
+    </insert>
+
+    <!-- Update statement -->
+    <update id="updateById" parameterType="com.wanji.indicators.entity.BaseLaneDetector">
+        UPDATE t_base_lane_detector
+        SET
+            geom = #{geom,jdbcType=LONGVARCHAR},
+            gid = #{gid,jdbcType=INTEGER},
+            xqid = #{xqid,jdbcType=VARCHAR},
+            flbm = #{flbm,jdbcType=VARCHAR},
+            lkid = #{lkid,jdbcType=VARCHAR},
+            ldid = #{ldid,jdbcType=VARCHAR},
+            qhdid = #{qhdid,jdbcType=VARCHAR},
+            cdid = #{cdid,jdbcType=VARCHAR}
+        WHERE id = #{id,jdbcType=INTEGER}
+    </update>
+
+    <!-- Delete statement -->
+    <delete id="deleteById" parameterType="int">
+        DELETE FROM t_base_lane_detector WHERE id = #{id,jdbcType=INTEGER}
+    </delete>
+
+    <!-- Select statement -->
+    <select id="selectById" parameterType="int" resultMap="BaseResultMap">
+        SELECT * FROM t_base_lane_detector WHERE id = #{id,jdbcType=INTEGER}
+    </select>
+
+    <!-- List all entries -->
+    <select id="selectSense" resultMap="BaseResultMap">
+        SELECT * FROM t_base_lane_detector
+        WHERE flbm = '感应检测器'
+    </select>
+
+</mapper>