1. 新增动态污染溯源的数据异常判断逻辑 2. 新增动态污染溯源websocket连接功能

feiyu02

2025-05-14 f1ea263462d1f3f10a886fb083536a38b03a1ceb

1. 新增动态污染溯源的数据异常判断逻辑
2. 新增动态污染溯源websocket连接功能

 src/main/kotlin/com/flightfeather/uav/UAVApplication.kt

@@ -23,7 +23,7 @@

    @Bean
    fun runner() = ApplicationRunner{
        underwaySocketServer.startWebSocketServer(9031, underwayProcessor)
        underwaySocketServer.startWebSocketServer(9031)
        underwaySocketServer.startUnderwayServer(9030, underwayProcessor)
        underwaySocketServer.startElectricServer(9009, electricProcessor)
    }

 src/main/kotlin/com/flightfeather/uav/biz/dataanalysis/BaseExceptionContinuous.kt

@@ -105,21 +105,23 @@
                }
                // åˆ¤æ–­ç›¸é‚»æ•°æ®æ˜¯å¦è¿žç»­å¹¶ä¸”是否满足异常判断
                if (!isContinue || needCut(it)) {
                    checkResult(s)
                    // æ•°æ®ä¸è¿žç»­æ—¶ï¼Œè®°å½•å¼‚常情况
                    if (it.eIndex - it.sIndex >= durationCount) {
                        it.refreshAfterCheckResult(data)
                    }
                    recordException(s, it, data)
//                    checkResult(s)
//                    if (it.eIndex - it.sIndex >= durationCount) {
//                        it.refreshAfterCheckResult(data)
//                    }
                } else {
                    if (hasException[f] == true) {
                        it.existException = true
                        it.exceptionData.add(data)
                    } else {
                        // å¼‚常不再重复出现时，记录异常情况
                        checkResult(s)
                        if (it.eIndex - it.sIndex >= durationCount) {
                            it.refreshAfterCheckResult(data)
                        }
                        recordException(s, it, data)
//                        checkResult(s)
//                        if (it.eIndex - it.sIndex >= durationCount) {
//                            it.refreshAfterCheckResult(data)
//                        }
                    }
                }
            }
@@ -131,6 +133,13 @@
        checkResult()
    }

    fun recordException(factor: FactorFilter.SelectedFactor, tag: Tag, data: BaseRealTimeData) {
        checkResult(factor)
        if (tag.eIndex - tag.sIndex >= durationCount) {
            tag.refreshAfterCheckResult(data)
        }
    }

    /**
     * æ£€æŸ¥è¿žç»­å¼‚常结束时，是否符合异常存储条件
     */
@@ -139,20 +148,12 @@
        if (factor != null && tag != null) {
            if (tag.existException && judgeExceptionCount(tag)) {
                onNewException(tag, factor)
//                tag.startData?.let {
//                    resultList.add(newResult(it, lastData, factor, tag.exceptionData))
//                }
//                tag.existException = false
            }
        } else {
            config.factorFilter.selectedList.forEach { f ->
                val tag1 = tagMap[f.main] ?: return@forEach
                if (tag1.existException && judgeExceptionCount(tag1)) {
                    onNewException(tag1, f)
//                    tag1.startData?.let {
//                        resultList.add(newResult(it, lastData, f, tag1.exceptionData))
//                    }
//                    tag1.existException = false
                }
            }
        }

 src/main/kotlin/com/flightfeather/uav/biz/sourcetrace/RealTimeAnalysisConfig.kt

@@ -25,6 +25,15 @@
    // çªå˜çŽ‡
    var mutationRate = .2

    // æº¯æºæœ‰æ•ˆæœ€å¤§é£Žé€Ÿï¼Œ5分钟不超过2公里的风速（暂定）
    var sourceTraceWindSpeedLimit = 6.7
    // æº¯æºæœ‰æ•ˆæœ€å¤§æ—¶é—´ï¼ˆå•ä½ï¼šåˆ†é’Ÿ)
    var sourceTraceTimeLimit = 5

    // æº¯æºæœ‰æ•ˆæœ€å¤§è·ç¦»ï¼ˆå•ä½ï¼šç±³ï¼‰
    var sourceTraceDistanceLimit = 2000

    // æº¯æºæœ‰æ•ˆæœ€å¤§é£Žé€Ÿï¼ˆå•ä½ï¼šç±³/秒），5分钟不超过2公里的风速（暂定）
    var sourceTraceWindSpeedLimit = 6.667

    // æº¯æºæ‰©æ•£åç§»è§’度（单位：度）
    var sourceTraceDegOffset = 30.0
}

 src/main/kotlin/com/flightfeather/uav/biz/sourcetrace/RealTimeExceptionAnalysisController.kt

@@ -4,68 +4,170 @@
import com.flightfeather.uav.biz.dataanalysis.BaseExceptionAnalysis
import com.flightfeather.uav.biz.sourcetrace.exceptiontype.RealTimeExceptionValueMutation
import com.flightfeather.uav.biz.sourcetrace.model.RealTimeExceptionResult
import com.flightfeather.uav.common.api2word.utils.JsonUtils
import com.flightfeather.uav.common.location.LocationRoadNearby
import com.flightfeather.uav.common.utils.GsonUtils
import com.flightfeather.uav.common.utils.MapUtil
import com.flightfeather.uav.domain.entity.BaseRealTimeData
import com.flightfeather.uav.domain.entity.SceneInfo
import com.flightfeather.uav.domain.entity.avg
import com.flightfeather.uav.domain.repository.RealTimeDataRep
import com.flightfeather.uav.domain.repository.SegmentInfoRep
import com.flightfeather.uav.domain.repository.SceneInfoRep
import com.flightfeather.uav.socket.eunm.FactorType
import com.flightfeather.uav.socket.handler.UnderwayWebSocketServerHandler
import com.google.gson.Gson
import com.flightfeather.uav.socket.sender.UnderwayWebSocketSender
import java.util.Timer
import java.util.TimerTask
import kotlin.math.PI

/**
 * å®žæ—¶èµ°èˆªæ±¡æŸ“溯源
 * @date 2025/5/8
 * @author feiyu02
 */
class RealTimeExceptionAnalysisController (
    private val realTimeDataRep: RealTimeDataRep,
    private val locationRoadNearby: LocationRoadNearby,
    private val segmentInfoRep: SegmentInfoRep,
    private val underwayWebSocketServerHandler: UnderwayWebSocketServerHandler,
    factorFilter: FactorFilter
){
    private var config:RealTimeAnalysisConfig = RealTimeAnalysisConfig(factorFilter)

class RealTimeExceptionAnalysisController {

    constructor(sceneInfoRep: SceneInfoRep, factorFilter: FactorFilter?) {
        this.sceneInfoRep = sceneInfoRep
        this.config = if (factorFilter != null)
            RealTimeAnalysisConfig(factorFilter)
        else
            RealTimeAnalysisConfig(
                FactorFilter.builder()
//                .withMain(FactorType.NO2)
                    .withMain(FactorType.CO)
//                .withMain(FactorType.H2S)
//                .withMain(FactorType.SO2)
//                .withMain(FactorType.O3)
                    .withMain(FactorType.PM25)
                    .withMain(FactorType.PM10)
                    .withMain(FactorType.VOC)
                    .create()
            )
        initTask(config)
    }

    constructor(sceneInfoRep: SceneInfoRep) : this(sceneInfoRep, null)


    private val sceneInfoRep: SceneInfoRep

    private val config: RealTimeAnalysisConfig

    private val taskList = mutableListOf<BaseExceptionAnalysis<RealTimeAnalysisConfig, RealTimeExceptionResult>>()

    private fun initTask(config: RealTimeAnalysisConfig) {
        taskList.clear()
        taskList.apply {
            add(RealTimeExceptionValueMutation(config){ exceptionCallback(it)})
        }
            add(
                RealTimeExceptionValueMutation(config) { exceptionCallback(it) }.also { it.init() }
            )
    }

    init {
        initTask(config)
    }

    // è®¡ç®—历史任务
    /**
     * è®¡ç®—新的一条实时走航数据
     */
    fun addOneData(data: BaseRealTimeData) {
        taskList

        // è®¡ç®—异常
        taskList.forEach { it.onNextData(data) }
        // é™å®šæ—¶é—´å†…没有新数据传入，则结束当前的计算
    }

    /**
     * è¶…时处理，有两种超时情况
     * 1. è¾ƒçŸ­æ—¶é—´å†…，主动结束连续当前异常判断
     * 2. è¾ƒé•¿æ—¶é—´å†…，进行初始化操作
     */
    private fun dealOnTimeout() {
        val timer = Timer(true)
        timer.schedule(object : TimerTask() {
            override fun run() {
                TODO("Not yet implemented")
            }
        }, 60 * 1000)
        timer.cancel()
    }

    // æ•°æ®çªå˜å¼‚常回调
    private fun exceptionCallback(ex: RealTimeExceptionResult) {
        if (sourceTrace(ex, config)) {
            underwayWebSocketServerHandler.broadcast(GsonUtils.gson.toJson(ex))
        }
        // æº¯æºæ±¡æŸ“源信息
        sourceTrace(ex, config)
        // å¹¿æ’­æ±¡æŸ“溯源异常结果
        UnderwayWebSocketSender.broadcast(GsonUtils.gson.toJson(ex))
    }

    private fun sourceTrace(ex: RealTimeExceptionResult, config: RealTimeAnalysisConfig):Boolean {
    /**
     * æ±¡æŸ“反向溯源
     */
    private fun sourceTrace(ex: RealTimeExceptionResult, config: RealTimeAnalysisConfig) {
        // è®¡ç®—异常数据均值
        val avgData = ex.dataList.avg()
        if (avgData.windSpeed!! > config.sourceTraceWindSpeedLimit) {
            return false
            return
        }

        // å–中间点作为反向溯源的起点
        val midData = ex.dataList[ex.dataList.size / 2]

//        avgData.longitude
//        avgData.latitude
//        avgData.windDirection
        return false
        // è®¡ç®—反向溯源区域
        val polygon = calSector(
            avgData.windSpeed!!.toDouble(),
            avgData.windDirection!!.toDouble(),
            midData.longitude!!.toDouble() to midData.latitude!!.toDouble(),
            config.sourceTraceTimeLimit,
            config.sourceTraceDegOffset
        )

        // æŒ‰ç…§åŒºåŸŸæ£€ç´¢å†…部污染源信息

        // 1. é¦–先按照四至范围从数据库初步筛选污染源
        val fb = MapUtil.calFourBoundaries(polygon)
        val sceneList = sceneInfoRep.findByCoordinateRange(fb)
        // 2. å†ç²¾ç¡®åˆ¤æ–­æ˜¯å¦åœ¨åå‘溯源区域多边形内部
        val result = mutableListOf<SceneInfo>()
        sceneList.forEach {
            // Fixme 2025.5.14: æ±¡æŸ“源的坐标是高德地图坐标系（火星坐标系），而走航数据是WGS84坐标系
            val point = MapUtil.gcj02ToWgs84(it!!.longitude.toDouble() to it.latitude.toDouble())
            if (MapUtil.isPointInPolygon(point, polygon)) {
                result.add(it)
            }
        }

        // æ›´æ–°ä¸­é—´ç‚¹ä¿¡æ¯
        ex.midData = avgData.apply {
            longitude = midData.longitude
            latitude = midData.latitude
        }
        // æ›´æ–°æº¯æºèŒƒå›´å†…的污染场景信息
        ex.relatedSceneList = result
    }

    /**
     * æ ¹æ®ä¸­å¿ƒç‚¹åæ ‡ã€é£Žå‘和风速，以及给定的夹角，计算以中心点按照风向风速和时长，向外扩散形成的扇形的点坐标
     * @param windSpeed é£Žé€Ÿï¼Œå•ä½ï¼šç±³/秒
     * @param windDir é£Žå‘，单位：度
     * @param center ä¸­å¿ƒç‚¹åæ ‡ç»çº¬åº¦
     * @param durationMin æ—¶é•¿ï¼Œå•ä½ï¼šåˆ†é’Ÿ
     * @param defaultDegOffset æ‰©æ•£åç§»è§’度
     * @return å¤šè¾¹å½¢é¡¶ç‚¹åæ ‡é›†åˆ
     */
    private fun calSector(
        windSpeed: Double, windDir: Double, center: Pair<Double, Double>, durationMin: Int,
        defaultDegOffset: Double = 30.0,
    ): List<Pair<Double, Double>> {

        val sDeg = windDir - defaultDegOffset
        val eDeg = windDir + defaultDegOffset
        val distance = windSpeed * durationMin * 60

        // å·¦ä¾§ï¼ˆé€†æ—¶é’ˆä¾§ï¼‰é¡¶ç‚¹
        val p1 = MapUtil.getPointByLen(center, distance, sDeg * PI / 180)
        // é£Žå‘反向顶点
        val p2 = MapUtil.getPointByLen(center, distance, windDir * PI / 180)
        // å³ä¾§ï¼ˆé¡ºæ—¶é’ˆä¾§ï¼‰é¡¶ç‚¹
        val p3 = MapUtil.getPointByLen(center, distance, eDeg * PI / 180)

        return listOf(center, p1, p2, p3)
    }

}

 src/main/kotlin/com/flightfeather/uav/biz/sourcetrace/exceptiontype/BaseRealTimeException.kt

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,73 @@
package com.flightfeather.uav.biz.sourcetrace.exceptiontype

import com.flightfeather.uav.biz.FactorFilter
import com.flightfeather.uav.biz.dataanalysis.BaseExceptionContinuous
import com.flightfeather.uav.biz.sourcetrace.RealTimeAnalysisConfig
import com.flightfeather.uav.biz.sourcetrace.model.RealTimeExceptionResult
import com.flightfeather.uav.domain.entity.BaseRealTimeData

// å¼‚常数据生成回调类
typealias NewExceptionCallback = (ex: RealTimeExceptionResult) -> Unit
/**
 * èµ°èˆªåŠ¨æ€æº¯æºå¼‚常判断
 * ç›¸æ¯”于其父类，修改了异常分组的逻辑，不再考虑异常需要连续出现，而是只通过数据采样时间是否连续以及是否需要主动截取分组来决定
 * å¦å¤–新增了异常实时播报的逻辑
 * @date 2025/5/13
 * @author feiyu02
 */
abstract class BaseRealTimeException(config: RealTimeAnalysisConfig) :
    BaseExceptionContinuous<RealTimeAnalysisConfig, RealTimeExceptionResult>(config) {

    constructor(config: RealTimeAnalysisConfig, callback: NewExceptionCallback) : this(config){
        this.callback = callback
    }

    var callback: NewExceptionCallback? = null

    override fun onNextData(data: BaseRealTimeData) {
        val isContinue = isContinuous(lastData, data)
        val hasException = judgeException(lastData, data)
        config.factorFilter.selectedList.forEach { s ->
            val f = s.main
            tagMap[f]?.let {
                it.eIndex++
                // èµ·å§‹æ•°æ®
                it.endData = data
                if (it.startData == null) {
                    it.refreshAfterCheckResult(data)
                }
                // åˆ¤æ–­ç›¸é‚»æ•°æ®æ˜¯å¦è¿žç»­æˆ–者是否满足自定义截取条件
                if (!isContinue || needCut(it)) {
                    // è®°å½•å¼‚常，结束异常的实时状态播报
                    recordException(s, it, data)
                } else {
                    if (hasException[f] == true) {
                        it.existException = true
                        it.exceptionData.add(data)
                    }
                    // ç§»é™¤äº†çˆ¶ç±»åŽŸæœ‰é€»è¾‘，改为当满足异常条件时，需要实时推送播报异常的状态变化，但不截取异常
                    checkResult(s)
                }
            }
        }
        lastData = data
    }

    override fun newResult(
        start: BaseRealTimeData,
        end: BaseRealTimeData?,
        factor: FactorFilter.SelectedFactor,
        exceptionData: List<BaseRealTimeData>,
    ): RealTimeExceptionResult {
        val eType = getExceptionType()
        return RealTimeExceptionResult(start, end, factor, exceptionData, eType)
    }

    override fun onNewException(tag: Tag, factor: FactorFilter.SelectedFactor) {
        super.onNewException(tag, factor)
        callback?.let { func ->
            val exc = resultList.last()
            func.invoke(exc)
        }
    }
}

 src/main/kotlin/com/flightfeather/uav/biz/sourcetrace/exceptiontype/RealTimeExceptionContinuous.kt

ÎļþÒÑɾ³ý

 src/main/kotlin/com/flightfeather/uav/biz/sourcetrace/exceptiontype/RealTimeExceptionValueMutation.kt

@@ -1,6 +1,5 @@
package com.flightfeather.uav.biz.sourcetrace.exceptiontype

import com.flightfeather.uav.biz.FactorFilter
import com.flightfeather.uav.biz.dataanalysis.model.ExceptionType
import com.flightfeather.uav.biz.sourcetrace.RealTimeAnalysisConfig
import com.flightfeather.uav.common.utils.MapUtil
@@ -15,7 +14,7 @@
 * @date 2025/5/13
 * @author feiyu02
 */
class RealTimeExceptionValueMutation : RealTimeExceptionContinuous {
class RealTimeExceptionValueMutation : BaseRealTimeException {

    constructor(config: RealTimeAnalysisConfig) : super(config)

@@ -53,7 +52,7 @@
    }

    override fun judgeExceptionCount(tag: Tag): Boolean {
        // é¦–个数据没有前一个数据参照，不算异常值，最后一个数据是判断结束的正常值，因此异常数据个数的计算下标为sIndex和eIndex
        // é¦–个数据没有前一个数据参照，不算异常值，最后一个数据是判断结束的正常值，因此异常数据个数的计算下标为sIndex和eIndex - 1
        val sIndex = tag.sIndex
        val eIndex = tag.eIndex - 1


 src/main/kotlin/com/flightfeather/uav/biz/sourcetrace/model/RealTimeExceptionResult.kt

@@ -7,6 +7,8 @@
import com.flightfeather.uav.domain.entity.BaseRealTimeData
import com.flightfeather.uav.domain.entity.SceneInfo
import com.flightfeather.uav.domain.entity.avg
import com.flightfeather.uav.lightshare.bean.DataVo
import com.flightfeather.uav.lightshare.eunm.ExceptionStatusType
import com.flightfeather.uav.socket.eunm.FactorType
import java.math.BigDecimal

@@ -16,6 +18,11 @@
 * @author feiyu02
 */
class RealTimeExceptionResult() : BaseExceptionResult() {

    // å¼‚常编号
    var guid: String? = null
    // å¼‚常的状态
    var status:Int = ExceptionStatusType.InProgress.value

    var deviceCode: String? = null

@@ -39,10 +46,16 @@

    // å¼‚常数据，头尾可能包含一定量的偏移
    var dataList: MutableList<BaseRealTimeData> = mutableListOf()
    var dataVoList: MutableList<DataVo> = mutableListOf()

    // ä¸­å¿ƒç‚¹ç»çº¬åº¦
    var longitude: BigDecimal? = null
    var latitude: BigDecimal? = null

    // ä¸­é—´æ•°æ®ç‚¹åæ ‡
    var midData: BaseRealTimeData? = null
//    var midLongitude: BigDecimal? = null
//    var midLatitude: BigDecimal? = null

    // æº¯æºä¼ä¸š
    var relatedSceneList: List<SceneInfo?>? = null
@@ -77,7 +90,10 @@
        min = s.second
        max = s.third

        exceptionData.forEach { dataList.add(it) }
        exceptionData.forEach {
            dataList.add(it)
            dataVoList.add(it.toDataVo())
        }
    }

    private fun dataSummary(

 src/main/kotlin/com/flightfeather/uav/common/utils/MapUtil.kt

@@ -1,14 +1,15 @@
package com.flightfeather.uav.common.utils

import kotlin.math.PI
import kotlin.math.asin
import kotlin.math.cos
import kotlin.math.sin
import kotlin.math.*

object MapUtil {

    private const val Ea = 6378137 //赤道半径
    private const val Eb = 6356725 //极半径

    // åæ ‡è½¬æ¢å‚æ•°
    const val a = 6378245.0; //长半轴
    const val ee = 0.00669342162296594323; //扁率/*** GCJ02 è½¬æ¢ä¸º WGS84* @param lng* @param lat* @returns {*[]}*/

    /**
     * æ ¹æ®åæ ‡ç‚¹ã€è·ç¦»å’Œè§’度，获取另一个坐标
@@ -93,6 +94,24 @@
    }

    /**
     * è®¡ç®—多边形的四至范围
     * @param polygon å¤šè¾¹å½¢åæ ‡ç‚¹æ•°ç»„
     * @return å››è‡³èŒƒå›´ï¼Œé¡ºåºä¸ºæœ€å°ç»åº¦ï¼Œæœ€å¤§ç»åº¦, æœ€å°çº¬åº¦ï¼Œæœ€å¤§çº¬åº¦
     */
    fun calFourBoundaries(polygon: List<Pair<Double, Double>>): List<Double> {
        // è®¡ç®—多边形顶点经度范围和纬度范围
        val xsSort = polygon.map { it.first }.sorted()
        val ysSort = polygon.map { it.second }.sorted()

        val xMin = xsSort[0]
        val yMin = ysSort[0]
        val xMax = xsSort[xsSort.lastIndex]
        val yMax = ysSort[ysSort.lastIndex]

        return listOf(xMin, xMax, yMin, yMax)
    }

    /**
     * åˆ¤æ–­åæ ‡ç‚¹æ˜¯å¦åœ¨å¤šè¾¹å½¢çš„四至范围内
     * @param point åæ ‡ç‚¹
     * @param polygon å¤šè¾¹å½¢åæ ‡ç‚¹æ•°ç»„
@@ -102,13 +121,19 @@
        val x = point.first
        val y = point.second
        // è®¡ç®—多边形顶点经度范围和纬度范围
        val xsSort = polygon.map { it.first }.sorted()
        val ysSort = polygon.map { it.second }.sorted()
        val fb = calFourBoundaries(polygon)
//        val xsSort = polygon.map { it.first }.sorted()
//        val ysSort = polygon.map { it.second }.sorted()
//
//        val xMin = xsSort[0]
//        val yMin = ysSort[0]
//        val xMax = xsSort[xsSort.lastIndex]
//        val yMax = ysSort[ysSort.lastIndex]

        val xMin = xsSort[0]
        val yMin = ysSort[0]
        val xMax = xsSort[xsSort.lastIndex]
        val yMax = ysSort[ysSort.lastIndex]
        val xMin = fb[0]
        val xMax = fb[1]
        val yMin = fb[2]
        val yMax = fb[3]

        return x >= xMin && x <= xMax && y >= yMin && y <= yMax
    }
@@ -177,4 +202,90 @@
        // è®¡ç®—是否在多边形内部
        return inPolygon(point, polygon)
    }

    /**
     * åˆ¤æ–­ç»çº¬åº¦æ˜¯å¦åœ¨å›½å†…
     * @return true: ç»çº¬åº¦ä¸åœ¨å›½å†…，false：经纬度在国内
     */
    fun outOfChina(point: Pair<Double, Double>): Boolean {
        val lng = point.first
        val lat = point.second
        return !(lng > 73.66 && lng < 135.05 && lat > 3.86 && lat < 53.55)
    }

    private fun transformLat(point: Pair<Double, Double>): Double {
        val lng = point.first
        val lat = point.second
        var ret = -100.0 +
                2.0 * lng +
                3.0 * lat +
                0.2 * lat * lat +
                0.1 * lng * lat +
                0.2 * sqrt(abs(lng))
        ret += ((20.0 * sin(6.0 * lng * PI) + 20.0 * sin(2.0 * lng * PI)) * 2.0) / 3.0
        ret += ((20.0 * sin(lat * PI) + 40.0 * sin((lat / 3.0) * PI)) * 2.0) / 3.0
        ret += ((160.0 * sin((lat / 12.0) * PI) + 320 * sin((lat * PI) / 30.0)) * 2.0) / 3.0
        return ret;
    }

    private fun transformLng(point: Pair<Double, Double>): Double {
        val lng = point.first
        val lat = point.second
        var ret = 300.0 +
                lng +
                2.0 * lat +
                0.1 * lng * lng +
                0.1 * lng * lat +
                0.1 * sqrt(abs(lng));
        ret += ((20.0 * sin(6.0 * lng * PI) + 20.0 * sin(2.0 * lng * PI)) * 2.0) / 3.0;
        ret += ((20.0 * sin(lng * PI) + 40.0 * sin((lng / 3.0) * PI)) * 2.0) / 3.0;
        ret += ((150.0 * sin((lng / 12.0) * PI) + 300.0 * sin((lng / 30.0) * PI)) * 2.0) / 3.0
        return ret;
    }

    /**
     * ç«æ˜Ÿåæ ‡ç³»è½¬WGS84坐标系
     */
    fun gcj02ToWgs84(point: Pair<Double, Double>): Pair<Double, Double> {
        if (outOfChina(point)) {
            return point;
        } else {
            val lng = point.first
            val lat = point.second
            var dlat = transformLat(lng - 105.0 to lat - 35.0);
            var dlng = transformLng(lng - 105.0 to lat - 35.0);
            val radlat = (lat / 180.0) * PI;
            var magic = sin(radlat);
            magic = 1 - ee * magic * magic;
            val sqrtmagic = sqrt(magic);
            dlat = (dlat * 180.0) / (((a * (1 - ee)) / (magic * sqrtmagic)) * PI);
            dlng = (dlng * 180.0) / ((a / sqrtmagic) * cos(radlat) * PI);
            val mglat = Math.round((lat * 2 - lat - dlat) * 1000000) / 1000000;
            val mglng = Math.round((lng * 2 - lng - dlng) * 1000000) / 1000000;
            return mglng.toDouble() to mglat.toDouble()
        }
    }

    /**
     * WGS84坐标系转火星坐标系
     */
    fun wgs84ToGcj02(point: Pair<Double, Double>): Pair<Double, Double> {
        if (outOfChina(point)) {
            return point
        } else {
            val lng = point.first
            val lat = point.second
            var dLat = transformLat(lng - 105.0 to lat - 35.0);
            var dLon = transformLng(lng - 105.0 to lat - 35.0);
            val radLat = (lat / 180.0) * PI;
            var magic = sin(radLat);
            magic = 1 - ee * magic * magic;
            val sqrtMagic = sqrt(magic);
            dLat = (dLat * 180.0) / (((a * (1 - ee)) / (magic * sqrtMagic)) * PI);
            dLon = (dLon * 180.0) / ((a / sqrtMagic) * cos(radLat) * PI);
            val mgLat = lat + dLat;
            val mgLon = lng + dLon;
            return mgLon to mgLat
        }
    }
}

 src/main/kotlin/com/flightfeather/uav/domain/entity/SceneInfo.java

@@ -28,9 +28,15 @@
    @Column(name = "Location")
    private String location;

    /**
     * ç»åº¦ï¼Œé«˜å¾·åœ°å›¾åæ ‡ç³»
     */
    @Column(name = "Longitude")
    private BigDecimal longitude;

    /**
     * çº¬åº¦ï¼Œé«˜å¾·åœ°å›¾åæ ‡ç³»
     */
    @Column(name = "Latitude")
    private BigDecimal latitude;


 src/main/kotlin/com/flightfeather/uav/domain/repository/impl/AirDataRepImpl.kt

@@ -132,8 +132,7 @@
        dataList.forEach {vo ->
            when (UWDeviceType.getType(vo.deviceCode)) {
                UWDeviceType.VEHICLE -> {
                    val d = RealTimeDataVehicle()
                    dataTransform(vo, d)
                    val d = vo.toBaseRealTimeData(RealTimeDataVehicle::class.java)
                    /***************************************************************************************************/
                    // FIXME: 2021/10/27 è½¦è½½ç›‘测部分因子量级调整
                    calibration(d, UWDeviceType.VEHICLE)
@@ -142,8 +141,7 @@
                    res.add(d)
                }
                UWDeviceType.UAV -> {
                    val d = RealTimeDataUav()
                    dataTransform(vo, d)
                    val d = vo.toBaseRealTimeData(RealTimeDataUav::class.java)
                    /***************************************************************************************************/
                    // FIXME: 2021/10/25 æ— äººæœºéƒ¨åˆ†å› å­é‡‡ç”¨è½¦è½½æ•°æ®å¡«å……，取最新的15分钟的数据
                    if (tmpVehicleDataList.isEmpty()) {
@@ -167,8 +165,7 @@
                    res.add(d)
                }
                UWDeviceType.GRID -> {
                    val d = RealTimeDataGrid()
                    dataTransform(vo, d)
                    val d = vo.toBaseRealTimeData(RealTimeDataGrid::class.java)
                    /**************************************************************************/
                    // FIXME: 2021/11/8 é’ˆå¯¹åŽ†å²ç½‘格化原始数据，进行临时校准处理
//                    val dTime = LocalDateTime.ofInstant(d.dataTime?.toInstant(), ZoneId.systemDefault())
@@ -267,38 +264,6 @@
                    }
                }
            }
        }
    }

    fun dataTransform(vo: DataVo, bean: BaseRealTimeData) {
        bean.apply {
            deviceCode = vo.deviceCode
            latitude = vo.lat?.toBigDecimal()
            longitude = vo.lng?.toBigDecimal()
            dataTime = SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(vo.time)
            createTime = Date()
            vo.values?.forEach {
                when (it.factorId?.toInt()) {
                    FactorType.NO2.value -> no2 = it.factorData?.toFloat()
                    FactorType.CO.value -> co = it.factorData?.toFloat()
                    FactorType.H2S.value -> h2s = it.factorData?.toFloat()
                    FactorType.SO2.value -> so2 = it.factorData?.toFloat()
                    FactorType.O3.value -> o3 = it.factorData?.toFloat()

                    FactorType.PM25.value -> pm25 = it.factorData?.toFloat()
                    FactorType.PM10.value -> pm10 = it.factorData?.toFloat()
                    FactorType.TEMPERATURE.value -> temperature = it.factorData?.toFloat()
                    FactorType.HUMIDITY.value -> humidity = it.factorData?.toFloat()
                    FactorType.VOC.value -> voc = it.factorData?.toFloat()

                    FactorType.NOI.value -> noi = it.factorData?.toFloat()
                    FactorType.VELOCITY.value -> velocity = it.factorData?.toFloat()
                    FactorType.WIND_SPEED.value -> windSpeed = it.factorData?.toFloat()
                    FactorType.WIND_DIRECTION.value -> windDirection = it.factorData?.toFloat()
                    FactorType.HEIGHT.value -> height = it.factorData?.toFloat()
                }
            }

        }
    }


 src/main/kotlin/com/flightfeather/uav/lightshare/bean/DataVo.kt

@@ -1,9 +1,12 @@
package com.flightfeather.uav.lightshare.bean

import com.fasterxml.jackson.annotation.JsonInclude
import com.flightfeather.uav.domain.entity.BaseRealTimeData
import com.flightfeather.uav.model.BaseMData
import com.flightfeather.uav.socket.bean.AirData
import com.flightfeather.uav.socket.eunm.FactorType
import java.text.SimpleDateFormat
import java.util.Date

/**
 * @author riku
@@ -20,7 +23,7 @@
        //经度
        var lng: Double? = null,
        //纬度
        var lat: Double? = null
    var lat: Double? = null,
) : BaseMData() {
        override fun getFactorData(type: FactorType): Double? {
                if (values == null) throw IllegalStateException(this.javaClass.name + ": ç›‘测数据数组为null")
@@ -62,4 +65,35 @@
                }
                return list.toTypedArray()
        }

    fun <T : BaseRealTimeData> toBaseRealTimeData(clz:Class<T>): T {
        return clz.newInstance().apply {
            deviceCode = this@DataVo.deviceCode
            latitude = this@DataVo.lat?.toBigDecimal()
            longitude = this@DataVo.lng?.toBigDecimal()
            dataTime = SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(this@DataVo.time)
            createTime = Date()
            this@DataVo.values?.forEach {
                when (it.factorId?.toInt()) {
                    FactorType.NO2.value -> no2 = it.factorData?.toFloat()
                    FactorType.CO.value -> co = it.factorData?.toFloat()
                    FactorType.H2S.value -> h2s = it.factorData?.toFloat()
                    FactorType.SO2.value -> so2 = it.factorData?.toFloat()
                    FactorType.O3.value -> o3 = it.factorData?.toFloat()

                    FactorType.PM25.value -> pm25 = it.factorData?.toFloat()
                    FactorType.PM10.value -> pm10 = it.factorData?.toFloat()
                    FactorType.TEMPERATURE.value -> temperature = it.factorData?.toFloat()
                    FactorType.HUMIDITY.value -> humidity = it.factorData?.toFloat()
                    FactorType.VOC.value -> voc = it.factorData?.toFloat()

                    FactorType.NOI.value -> noi = it.factorData?.toFloat()
                    FactorType.VELOCITY.value -> velocity = it.factorData?.toFloat()
                    FactorType.WIND_SPEED.value -> windSpeed = it.factorData?.toFloat()
                    FactorType.WIND_DIRECTION.value -> windDirection = it.factorData?.toFloat()
                    FactorType.HEIGHT.value -> height = it.factorData?.toFloat()
                }
            }
        }
    }
}

 src/main/kotlin/com/flightfeather/uav/lightshare/eunm/ExceptionStatusType.kt

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,11 @@
package com.flightfeather.uav.lightshare.eunm

/**
 * å¼‚常状态
 * @date 2025/5/14
 * @author feiyu02
 */
enum class ExceptionStatusType(val value: Int, val des: String) {
    InProgress(1, "持续中"),
    Ended(2, "已结束"),
}

 src/main/kotlin/com/flightfeather/uav/socket/UnderwaySocketServer.kt

@@ -1,5 +1,6 @@
package com.flightfeather.uav.socket

import com.flightfeather.uav.domain.repository.SceneInfoRep
import com.flightfeather.uav.socket.handler.ServerHandler
import com.flightfeather.uav.socket.handler.UnderwayWebSocketServerHandler
import com.flightfeather.uav.socket.processor.BaseProcessor
@@ -25,7 +26,7 @@
 * *******************************************************************************/
@Component
class UnderwaySocketServer(
    private val underwayWebSocketServerHandler:UnderwayWebSocketServerHandler
    private val sceneInfoRep: SceneInfoRep
) {

    private val bossGroup = NioEventLoopGroup()
@@ -39,8 +40,8 @@
        electricServer(processor)?.bind(port)?.sync()
    }

    fun startWebSocketServer(port: Int, processor: BaseProcessor) {
        webSocketServer(processor)?.bind(port)?.sync()
    fun startWebSocketServer(port: Int) {
        webSocketServer()?.bind(port)?.sync()
    }

    fun stopServer() {
@@ -92,13 +93,13 @@
    /**
     * å¤šå‚数走航服务端
     */
    private fun webSocketServer(processor: BaseProcessor):ServerBootstrap? = newServer(object : ChannelInitializer<NioSocketChannel>() {
    private fun webSocketServer():ServerBootstrap? = newServer(object : ChannelInitializer<NioSocketChannel>() {
        override fun initChannel(p0: NioSocketChannel?) {
            p0?.pipeline()
                ?.addLast(HttpServerCodec())
                ?.addLast(HttpObjectAggregator(65535))
                ?.addLast(WebSocketServerProtocolHandler("/ws"))
                ?.addLast(underwayWebSocketServerHandler)
                ?.addLast(UnderwayWebSocketServerHandler(sceneInfoRep))
        }
    })
}

 src/main/kotlin/com/flightfeather/uav/socket/handler/BaseHandler.kt

@@ -15,12 +15,10 @@
    abstract var tag: String

    override fun channelRegistered(ctx: ChannelHandlerContext?) {
        super.channelRegistered(ctx)
        println("------【${tag}】IP连接：[ip:${ctx?.channel()?.remoteAddress()}] ${
            SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(
                Date()
            )}")
//        ctx?.fireChannelActive()
    }

    override fun channelActive(ctx: ChannelHandlerContext?) {
@@ -28,19 +26,13 @@
            SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(
                Date()
            )}")
        super.channelActive(ctx)
    }

    override fun channelRead(ctx: ChannelHandlerContext?, msg: Any?) {
        super.channelRead(ctx, msg)
        println("------【${tag}】收到的原始数据：[ip:${ctx?.channel()?.remoteAddress()}] ${
            SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(
                Date()
            )}")
    }

    override fun channelReadComplete(ctx: ChannelHandlerContext?) {
        super.channelReadComplete(ctx)
    }

    override fun channelInactive(ctx: ChannelHandlerContext?) {
@@ -48,7 +40,6 @@
            SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(
                Date()
            )}")
        super.channelInactive(ctx)
    }

    @Deprecated("Deprecated in Java")

 src/main/kotlin/com/flightfeather/uav/socket/handler/UnderwayWebSocketServerHandler.kt

@@ -1,5 +1,13 @@
package com.flightfeather.uav.socket.handler

import com.flightfeather.uav.biz.sourcetrace.RealTimeExceptionAnalysisController
import com.flightfeather.uav.common.api2word.utils.JsonUtils
import com.flightfeather.uav.common.utils.GsonUtils
import com.flightfeather.uav.domain.entity.BaseRealTimeData
import com.flightfeather.uav.domain.repository.SceneInfoRep
import com.flightfeather.uav.lightshare.bean.DataVo
import com.flightfeather.uav.socket.sender.UnderwayWebSocketSender
import com.google.gson.JsonSyntaxException
import io.netty.channel.ChannelHandlerContext
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame
import org.springframework.stereotype.Component
@@ -9,28 +17,39 @@
 * @date 2025/5/13
 * @author feiyu02
 */
@Component
class UnderwayWebSocketServerHandler : BaseHandler() {
class UnderwayWebSocketServerHandler(sceneInfoRep: SceneInfoRep) : BaseHandler() {

    private val sessionPool = mutableMapOf<String?, ChannelHandlerContext?>()
    private val realTimeExceptionAnalysisController = RealTimeExceptionAnalysisController(sceneInfoRep)

    override var tag: String = "UAV-WS"

    override fun channelRegistered(ctx: ChannelHandlerContext?) {
        super.channelRegistered(ctx)
        // å°†è¿žæŽ¥å­˜å‚¨
        if (!sessionPool.containsKey(ctx?.name())) {
            sessionPool[ctx?.name()] = ctx
        }
        UnderwayWebSocketSender.saveSession(ctx)
    }

    override fun channelRead(ctx: ChannelHandlerContext?, msg: Any?) {
        super.channelRead(ctx, msg)

        when (msg) {
            is TextWebSocketFrame->{
                println(msg.text())
                ctx?.channel()?.writeAndFlush(msg)
                val msgTxt = msg.text()

                println(msgTxt)
//                ctx?.channel()?.writeAndFlush(msg)

                // Test
                try {
                    val data = GsonUtils.parserJsonToArrayBeans(msgTxt, DataVo::class.java)
                    data.forEach {
                        realTimeExceptionAnalysisController.addOneData(
                            it.toBaseRealTimeData(BaseRealTimeData::class.java)
                        )
                    }
                } catch (e: Exception) {
//                    ctx?.channel()?.writeAndFlush(TextWebSocketFrame("当前为测试状态，传输的数据不是走航数据格式"))
                    println("当前为测试状态，传输的数据不是走航数据格式")
                }
            }
        }
    }
@@ -38,18 +57,6 @@
    override fun channelInactive(ctx: ChannelHandlerContext?) {
        super.channelInactive(ctx)
        // å°†è¿žæŽ¥ç§»é™¤
        if (sessionPool.containsKey(ctx?.name())) {
            sessionPool.remove(ctx?.name())
        }
    }

    fun send() {

    }

    fun broadcast(msg: String) {
        sessionPool.forEach { t, u ->
            u?.channel()?.writeAndFlush(TextWebSocketFrame(msg))
        }
        UnderwayWebSocketSender.removeSession(ctx)
    }
}

 src/main/kotlin/com/flightfeather/uav/socket/processor/UnderwayProcessor.kt

@@ -7,6 +7,7 @@
import com.flightfeather.uav.model.epw.EPWDataPrep
import com.flightfeather.uav.domain.repository.AirDataRep
import com.flightfeather.uav.domain.repository.RealTimeDataRep
import com.flightfeather.uav.domain.repository.SceneInfoRep
import com.flightfeather.uav.domain.repository.SegmentInfoRep
import com.flightfeather.uav.socket.bean.AirDataPackage
import com.flightfeather.uav.socket.decoder.AirDataDecoder
@@ -31,10 +32,7 @@
@Component
class UnderwayProcessor(
    private val airDataRep: AirDataRep,
    private val realTimeDataRep: RealTimeDataRep,
    private val locationRoadNearby: LocationRoadNearby,
    private val segmentInfoRep: SegmentInfoRep,
    private val underwayWebSocketServerHandler: UnderwayWebSocketServerHandler,
    private val sceneInfoRep: SceneInfoRep,
) : BaseProcessor() {

    companion object {
@@ -48,23 +46,7 @@
    private val dataProcessMap = mutableMapOf<String?, EPWDataPrep>()

    // å®žæ—¶èµ°èˆªæ±¡æŸ“溯源处理器
    private val realTimeExceptionAnalysisController =
        RealTimeExceptionAnalysisController(
            realTimeDataRep,
            locationRoadNearby,
            segmentInfoRep,
            underwayWebSocketServerHandler,
            FactorFilter.builder()
//                .withMain(FactorType.NO2)
                .withMain(FactorType.CO)
//                .withMain(FactorType.H2S)
//                .withMain(FactorType.SO2)
//                .withMain(FactorType.O3)
                .withMain(FactorType.PM25)
                .withMain(FactorType.PM10)
                .withMain(FactorType.VOC)
                .create()
        )
    private val realTimeExceptionAnalysisMap = mutableMapOf<String?, RealTimeExceptionAnalysisController>()

    override var tag: String = "走航监测"

@@ -77,8 +59,12 @@
            deviceSession.saveDevice(packageData.deviceCode, ctx)
            saveToTxt(msg)
            saveToDataBase(packageData)?.takeIf { it.isNotEmpty() }?.get(0)?.let {
                // æ¯å°è®¾å¤‡æœ‰å„自单独的异常数据处理器
                if (!realTimeExceptionAnalysisMap.containsKey(it.deviceCode)) {
                    realTimeExceptionAnalysisMap[it.deviceCode] = RealTimeExceptionAnalysisController(sceneInfoRep)
                }
                // å°†èµ°èˆªæ•°æ®ä¼ å…¥å¼‚常处理器
                realTimeExceptionAnalysisController.addOneData(it)
                realTimeExceptionAnalysisMap[it.deviceCode]?.addOneData(it)
            }

        } else {

 src/main/kotlin/com/flightfeather/uav/socket/sender/UnderwayWebSocketSender.kt

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,36 @@
package com.flightfeather.uav.socket.sender

import io.netty.channel.ChannelHandlerContext
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame

/**
 *
 * @date 2025/5/14
 * @author feiyu02
 */
object UnderwayWebSocketSender {

    private val sessionPool = mutableMapOf<String?, ChannelHandlerContext?>()

    fun saveSession(ctx: ChannelHandlerContext?) {
        if (!sessionPool.containsKey(ctx?.name())) {
            sessionPool[ctx?.name()] = ctx
        }
    }

    fun removeSession(ctx: ChannelHandlerContext?) {
        if (sessionPool.containsKey(ctx?.name())) {
            sessionPool.remove(ctx?.name())
        }
    }

    fun send() {

    }

    fun broadcast(msg: String) {
        sessionPool.forEach { (t, u) ->
            u?.channel()?.writeAndFlush(TextWebSocketFrame(msg))
        }
    }
}

 src/test/kotlin/com/flightfeather/uav/lightshare/service/impl/RealTimeDataServiceImplTest.kt

@@ -23,8 +23,6 @@
    @Autowired
    lateinit var realTimeDataVehicleMapper: RealTimeDataVehicleMapper

    @Autowired
    lateinit var airDataRepository: AirDataRepImpl

    @Test
    fun outToExcel() {
@@ -153,8 +151,7 @@
            }
            println("当前页数：$page")
            res.data?.forEach {vo ->
                val d = RealTimeDataVehicle()
                airDataRepository.dataTransform(vo, d)
                val d = vo.toBaseRealTimeData(RealTimeDataVehicle::class.java)
                realTimeDataVehicleMapper.insert(d)
                count++
            }

 src/test/kotlin/com/flightfeather/uav/socket/UnderwayProcessorTest.kt

@@ -2,14 +2,21 @@

import com.flightfeather.uav.socket.processor.UnderwayProcessor
import org.junit.Test
import org.junit.runner.RunWith
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.boot.test.context.SpringBootTest
import org.springframework.test.context.junit4.SpringRunner

/**
 * @author riku
 * Date: 2019/9/16
 */
@RunWith(SpringRunner::class)
@SpringBootTest
class UnderwayProcessorTest {

    private val messageManager = UnderwayProcessor()
    @Autowired
    lateinit var messageManager: UnderwayProcessor

    @Test
    fun bccCheck() {