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

feiyu02

2025-05-13 1571cd0f137ced4345fa8785e166a29dc31b6ad1

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

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

@@ -18,10 +18,14 @@
    @Autowired
    lateinit var electricProcessor: ElectricProcessor

    @Autowired
    lateinit var underwaySocketServer:UnderwaySocketServer

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


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

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

import com.flightfeather.uav.biz.FactorFilter

/**
 * æ•°æ®åˆ†æžé…ç½®å‚数基类
 * @date 2025/5/13
 * @author feiyu02
 */
abstract class BaseAnalysisConfig(
    // å› å­ç­›é€‰
    val factorFilter: FactorFilter,
) {
}

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

@@ -1,93 +1,17 @@
package com.flightfeather.uav.biz.dataanalysis

import com.flightfeather.uav.biz.FactorFilter
import com.flightfeather.uav.biz.dataanalysis.model.DataAnalysisConfig
import com.flightfeather.uav.biz.dataanalysis.model.ExceptionResult
import com.flightfeather.uav.biz.dataanalysis.model.ExceptionType
import com.flightfeather.uav.common.utils.DateUtil
import com.flightfeather.uav.domain.entity.BaseRealTimeData
import com.flightfeather.uav.domain.entity.avg
import com.flightfeather.uav.socket.eunm.FactorType
import java.time.Duration

/**
 * ç›‘测数据异常分析基类
 */
abstract class BaseExceptionAnalysis(config: DataAnalysisConfig) :
    BaseDataAnalysis<BaseRealTimeData, DataAnalysisConfig, ExceptionResult>(config) {
abstract class BaseExceptionAnalysis<V : BaseAnalysisConfig, Y : BaseExceptionResult>(config: V) :
    BaseDataAnalysis<BaseRealTimeData, V, Y>(config) {

    /**
     * ç¡®å®šå¼‚常类型
     */
    abstract fun getExceptionType(): ExceptionType

    /**
     * åˆ¤æ–­ç›¸é‚»æ•°æ®æ˜¯å¦è¿žç»­
     */
    open fun isContinuous(d1: BaseRealTimeData?, d2: BaseRealTimeData?): Boolean {
        if (d1 == null || d2 == null) return true

        val t1 = d1.dataTime
        val t2 = d2.dataTime
        return Duration.between(t1?.toInstant(), t2?.toInstant()).toMillis() <= (20 * 1000)
    }

    /**
     * ç”Ÿæˆä¸€æ¡å¼‚常分析结果
     */
    open fun newResult(
        start: BaseRealTimeData, end: BaseRealTimeData?, factor: FactorFilter.SelectedFactor,
        exceptionData: List<BaseRealTimeData>,
    ): ExceptionResult {
        val eType = getExceptionType()
        return ExceptionResult().apply {
            missionCode = config.mission.missionCode
            deviceCode = start.deviceCode
            exception = eType.des
            exceptionType = eType.value
            factorId = factor.main.value
            factorName = factor.main.des
            subFactorId = factor.subs.map { it.value }
            subFactorName = factor.subs.map { it.des }
            selectedFactor = factor
            startDate = start.dataTime
            endDate = end?.dataTime
            startTime = DateUtil.instance.dateToString(start.dataTime, DateUtil.DateStyle.HH_MM_SS)
            endTime = DateUtil.instance.dateToString(end?.dataTime, DateUtil.DateStyle.HH_MM_SS) ?: startTime
            startData = start.getByFactorType(factor.main)
            endData = end?.getByFactorType(factor.main) ?: startData

            val avgData = exceptionData.avg()
            // æ±‚取污染数据的中心坐标
            longitude = avgData.longitude
            latitude = avgData.latitude
            // æ±‚取主污染因子的均值和范围
            val s = dataSummary(exceptionData, factor.main)
            avg = s.first
            min = s.second
            max = s.third

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

    private fun dataSummary(exceptionData: List<BaseRealTimeData?>, factorType: FactorType): Triple<Float, Float, Float> {
        var min = -1f
        var max = -1f
        var total = 0f
        var count = 0
        exceptionData.forEach {
            val value = it?.getByFactorType(factorType) ?: return@forEach
            if (min == -1f || min > value) {
                min = value
            }
            if (max == -1f || max < value) {
                max = value
            }
            total += value
            count++
        }
        val avg = if (count == 0) 0f else total / count
        return Triple(avg, min, max)
    }
}

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

@@ -1,14 +1,15 @@
package com.flightfeather.uav.biz.dataanalysis

import com.flightfeather.uav.biz.FactorFilter
import com.flightfeather.uav.biz.dataanalysis.model.DataAnalysisConfig
import com.flightfeather.uav.domain.entity.BaseRealTimeData
import com.flightfeather.uav.socket.eunm.FactorType
import java.time.Duration

/**
 * è¿žç»­ç±»åž‹çš„异常分析基类,适用于当前数据与相邻数据之间有关联关系的情况
 */
abstract class BaseExceptionContinuous(config: DataAnalysisConfig) : BaseExceptionAnalysis(config) {
abstract class BaseExceptionContinuous<V : BaseAnalysisConfig, Y : BaseExceptionResult>(config: V) :
    BaseExceptionAnalysis<V, Y>(config) {

    companion object {
        // è®°å½•å¼‚常数据段时，分别向起始前和末尾后额外记录的数据个数偏移量
@@ -38,30 +39,28 @@
            sIndex = eIndex
            startData = data
            exceptionData.clear()
            exceptionData.add(data)
//            exceptionData.add(data)
        }
    }

    protected val tagMap = mutableMapOf<FactorType, Tag>()

//    // èµ·å§‹æ•°æ®ä¸‹æ ‡
//    protected var sIndex = mutableListOf<Int>()
//
//    // èµ·å§‹æ•°æ®å¯¹è±¡
//    protected var startData = mutableListOf<BaseRealTimeData?>()
//
//    // æœ«å°¾æ•°æ®ä¸‹æ ‡
//    protected var eIndex = mutableListOf<Int>()
//
//    // å¼‚常数据段
//    protected var exceptionData = mutableListOf<MutableList<BaseRealTimeData>>()

//    protected var existException = mutableListOf<Boolean>()

    // èµ·å§‹æ•°æ®ä¸Žæœ«å°¾æ•°æ®é—´éš”
    open var durationCount = 1

    // æœ«å°¾æ•°æ®å¯¹è±¡
    protected var lastData: BaseRealTimeData? = null

    /**
     * åˆ¤æ–­ç›¸é‚»æ•°æ®æ˜¯å¦è¿žç»­
     */
    open fun isContinuous(d1: BaseRealTimeData?, d2: BaseRealTimeData?): Boolean {
        if (d1 == null || d2 == null) return true

        val t1 = d1.dataTime
        val t2 = d2.dataTime
        return Duration.between(t1?.toInstant(), t2?.toInstant()).toMillis() <= (20 * 1000)
    }

    /**
     * åˆ¤æ–­æ˜¯å¦æ»¡è¶³å¼‚常条件
@@ -69,22 +68,23 @@
    abstract fun judgeException(p: BaseRealTimeData?, n: BaseRealTimeData): MutableMap<FactorType, Boolean>

    /**
     * åˆ¤æ–­å¼‚常出现的连续时长是否满足条件
     * @param sIndex
     * @param eIndex
     * åˆ¤æ–­å¼‚常出现的连续个数是否满足条件
     * @param tag å¼‚常数据对象
     */
    abstract fun judgeDuration(sIndex: Int, eIndex: Int): Boolean
    abstract fun judgeExceptionCount(tag: Tag): Boolean

    /**
     * å¼‚常数据的截取判断
     * æ˜¯å¦éœ€è¦é™åˆ¶ä¸€ç»„异常数据的长度
     * @return é»˜è®¤ä¸éœ€è¦æˆªå–
     */
    open fun needCut(tag: Tag): Boolean {
        return false
    }

    override fun init() {
        super.init()
        lastData = null
//        repeat(config.factorCount) {
//            startData.add(null)
//            sIndex.add(0)
//            eIndex.add(-1)
//            existException.add(false)
//            exceptionData.add(mutableListOf())
//        }
        tagMap.clear()
        config.factorFilter.mainList().forEach {f->
            tagMap[f] = Tag()
@@ -99,12 +99,12 @@
            tagMap[f]?.let {
                it.eIndex++
                // èµ·å§‹æ•°æ®
                it.endData = lastData
                if (it.endData == null) {
                it.endData = data
                if (it.startData == null) {
                    it.refreshAfterCheckResult(data)
                }
                // åˆ¤æ–­ç›¸é‚»æ•°æ®æ˜¯å¦è¿žç»­å¹¶ä¸”是否满足异常判断
                if (!isContinue) {
                if (!isContinue || needCut(it)) {
                    checkResult(s)
                    // æ•°æ®ä¸è¿žç»­æ—¶ï¼Œè®°å½•å¼‚常情况
                    if (it.eIndex - it.sIndex >= durationCount) {
@@ -131,43 +131,51 @@
        checkResult()
    }

//    fun refreshAfterCheckResult(i:Int, data: BaseRealTimeData) {
//        sIndex[i] = eIndex[i]
//        startData[i] = data
//        exceptionData[i].clear()
//        exceptionData[i].add(data)
//    }

    /**
     * æ£€æŸ¥è¿žç»­å¼‚常结束时，是否符合异常存储条件
     */
    open fun checkResult(factor: FactorFilter.SelectedFactor? = null) {
        val tag = tagMap[factor?.main]
        if (factor != null && tag != null) {
            if (tag.existException && judgeDuration(tag.sIndex, tag.eIndex - 1)) {
            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
                }
            }
        }
    }

    /**
     * æ–°å¢žä¸€æ¡å¼‚常
     */
    open fun onNewException(tag:Tag, factor: FactorFilter.SelectedFactor) {
                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 && judgeDuration(tag1.sIndex, tag1.eIndex - 1)) {
                    tag1.startData?.let {
                        resultList.add(newResult(it, lastData, f, tag1.exceptionData))
                    }
                    tag1.existException = false
                }
            }
//            repeat(config.factorCount) { i ->
//                if (existException[i] && judgeDuration(sIndex[i], eIndex[i])) {
//                    startData[i]?.let {
//                        resultList.add(newResult(it, lastData, i, exceptionData[i]))
//                    }
//                    existException[i] = false
//                }
//            }
        }
    }

    /**
     * ç”Ÿæˆä¸€æ¡å¼‚常分析结果
     */
    abstract fun newResult(
        start: BaseRealTimeData,
        end: BaseRealTimeData?,
        factor: FactorFilter.SelectedFactor,
        exceptionData: List<BaseRealTimeData>,
    ): Y

}

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

@@ -1,12 +1,12 @@
package com.flightfeather.uav.biz.dataanalysis

import com.flightfeather.uav.biz.dataanalysis.model.DataAnalysisConfig
import com.flightfeather.uav.domain.entity.BaseRealTimeData

/**
 * è¿žç»­ç±»åž‹çš„异常分析基类,区别于父类的地方在于此种异常只和单个数据本身有关,与相邻数据无关
 */
abstract class BaseExceptionContinuousSingle(config: DataAnalysisConfig) : BaseExceptionContinuous(config) {
abstract class BaseExceptionContinuousSingle<V : BaseAnalysisConfig, Y : BaseExceptionResult>(config: V) :
    BaseExceptionContinuous<V, Y>(config) {

    override fun onNextData(data: BaseRealTimeData) {
        val isContinue = isContinuous(lastData, data)
@@ -19,12 +19,12 @@
                    it.startData = data
                }
                // åˆ¤æ–­ç›¸é‚»æ•°æ®æ˜¯å¦è¿žç»­å¹¶ä¸”是否满足异常判断
                if (!isContinue) {
                if (!isContinue || needCut(it)) {
                    checkResult(s)
                    it.refreshAfterCheckResult(data)
                } else {
                    if (hasException[f] == true) {
                        // ä¿®æ”¹äº†èµ·å§‹æ•°æ®çš„位置,变更为出现异常的该值,而不是原来的出现异常的数据的前一个值
                        // ä¿®æ”¹äº†èµ·å§‹æ•°æ®çš„位置,变更为出现异常的该值,而不是原来的出现异常数据的前一个值
                        if (!it.existException) {
                            it.sIndex = it.eIndex
                            it.startData = data
@@ -38,29 +38,6 @@
                }
            }
        }
//        repeat(config.factorCount) { i ->
//            eIndex[i]++
//            if (lastData == null) {
//                startData[i] = data
//            }
//            // åˆ¤æ–­ç›¸é‚»æ•°æ®æ˜¯å¦è¿žç»­å¹¶ä¸”是否满足异常判断
//            if (!isContinue) {
//                checkResult()
//                sIndex[i] = eIndex[i]
//                startData[i] = data
//            } else {
//                if (hasException[i]) {
//                    // ä¿®æ”¹äº†èµ·å§‹æ•°æ®çš„位置,变更为出现异常的该值,而不是原来的出现异常的数据的前一个值
//                    if (!existException[i]) {
//                        sIndex[i] = eIndex[i]
//                        startData[i] = data
//                    }
//                    existException[i] = true
//                } else {
//                    checkResult()
//                }
//            }
//        }
        lastData = data
    }
}

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

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

/**
 * å¼‚常结果基类
 * @date 2025/5/13
 * @author feiyu02
 */
abstract class BaseExceptionResult {
}

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

@@ -25,7 +25,7 @@

    var running = false

    private val taskList = mutableListOf<BaseExceptionAnalysis>()
    private val taskList = mutableListOf<BaseExceptionAnalysis<DataAnalysisConfig, ExceptionResult>>()

    private fun initTask(config: DataAnalysisConfig) {
        taskList.clear()

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

ÎļþÒÑɾ³ý

 src/main/kotlin/com/flightfeather/uav/biz/dataanalysis/exceptiontype/ExceptionContinuous.kt

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

import com.flightfeather.uav.biz.FactorFilter
import com.flightfeather.uav.biz.dataanalysis.BaseExceptionContinuous
import com.flightfeather.uav.biz.dataanalysis.model.DataAnalysisConfig
import com.flightfeather.uav.biz.dataanalysis.model.ExceptionResult
import com.flightfeather.uav.domain.entity.BaseRealTimeData

/**
 * è¿žç»­ç±»åž‹çš„异常分析基类,适用于当前数据与相邻数据之间有关联关系的情况
 * @date 2025/5/13
 * @author feiyu02
 */
abstract class ExceptionContinuous(config: DataAnalysisConfig) :
    BaseExceptionContinuous<DataAnalysisConfig, ExceptionResult>(config) {

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

 src/main/kotlin/com/flightfeather/uav/biz/dataanalysis/exceptiontype/ExceptionDataExceed.kt

@@ -1,7 +1,9 @@
package com.flightfeather.uav.biz.dataanalysis.exceptiontype

import com.flightfeather.uav.biz.FactorFilter
import com.flightfeather.uav.biz.dataanalysis.BaseExceptionContinuousSingle
import com.flightfeather.uav.biz.dataanalysis.model.DataAnalysisConfig
import com.flightfeather.uav.biz.dataanalysis.model.ExceptionResult
import com.flightfeather.uav.biz.dataanalysis.model.ExceptionType
import com.flightfeather.uav.domain.entity.BaseRealTimeData
import com.flightfeather.uav.socket.eunm.FactorType
@@ -10,7 +12,8 @@
 * æ•°æ®è¶…标异常分析
 */
@Deprecated("原超标判定逻辑设定的超标值有漏洞，")
class ExceptionDataExceed(config: DataAnalysisConfig) : BaseExceptionContinuousSingle(config) {
class ExceptionDataExceed(config: DataAnalysisConfig) :
    BaseExceptionContinuousSingle<DataAnalysisConfig, ExceptionResult>(config) {

    override fun getExceptionType(): ExceptionType = ExceptionType.TYPE2

@@ -29,7 +32,17 @@
        return res
    }

    override fun judgeDuration(sIndex: Int, eIndex: Int): Boolean {
    override fun judgeExceptionCount(tag: Tag): Boolean {
        return true
    }

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

 src/main/kotlin/com/flightfeather/uav/biz/dataanalysis/exceptiontype/ExceptionSlideAverage.kt

@@ -3,6 +3,7 @@
import com.flightfeather.uav.biz.FactorFilter
import com.flightfeather.uav.biz.dataanalysis.BaseExceptionAnalysis
import com.flightfeather.uav.biz.dataanalysis.model.DataAnalysisConfig
import com.flightfeather.uav.biz.dataanalysis.model.ExceptionResult
import com.flightfeather.uav.biz.dataanalysis.model.ExceptionType
import com.flightfeather.uav.domain.entity.BaseRealTimeData
import com.flightfeather.uav.socket.eunm.FactorType
@@ -11,7 +12,8 @@
/**
 * æ»‘动平均值突变异常
 */
class ExceptionSlideAverage(config: DataAnalysisConfig) : BaseExceptionAnalysis(config) {
class ExceptionSlideAverage(config: DataAnalysisConfig) :
    BaseExceptionAnalysis<DataAnalysisConfig, ExceptionResult>(config) {

    private val historyDataList = mutableListOf<BaseRealTimeData>()
    private val tempDataList = mutableListOf<BaseRealTimeData>()
@@ -198,4 +200,14 @@
        }

    }

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

 src/main/kotlin/com/flightfeather/uav/biz/dataanalysis/exceptiontype/ExceptionValueMutation.kt

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

import com.flightfeather.uav.biz.dataanalysis.BaseExceptionContinuous
import com.flightfeather.uav.biz.dataanalysis.model.DataAnalysisConfig
import com.flightfeather.uav.biz.dataanalysis.model.ExceptionType
import com.flightfeather.uav.domain.entity.BaseRealTimeData
@@ -10,7 +9,7 @@
/**
 * é‡çº§çªå˜å¼‚常分析
 */
class ExceptionValueMutation(config: DataAnalysisConfig) : BaseExceptionContinuous(config) {
class ExceptionValueMutation(config: DataAnalysisConfig) : ExceptionContinuous(config) {

    /**
     * æœ¬å¼‚常的连续发生次数会根据异常的程度变化
@@ -40,7 +39,11 @@
        return res
    }

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

        val b1 = special && (eIndex - sIndex) >= (config.mutationNum / 2)
        val b2 = (eIndex - sIndex) >= config.mutationNum
        special = false

 src/main/kotlin/com/flightfeather/uav/biz/dataanalysis/model/DataAnalysisConfig.kt

@@ -1,29 +1,33 @@
package com.flightfeather.uav.biz.dataanalysis.model

import com.flightfeather.uav.biz.FactorFilter
import com.flightfeather.uav.biz.dataanalysis.BaseAnalysisConfig
import com.flightfeather.uav.domain.entity.Mission

/**
 * æ•°æ®åˆ†æžé…ç½®å‚æ•°
 */
data class DataAnalysisConfig(
class DataAnalysisConfig(
    // èµ°èˆªä»»åŠ¡ä¿¡æ¯
    val mission: Mission,
    // æ•°æ®å¼‚常配置
    val exceptionSetting: ExceptionSetting,
    // å› å­ç­›é€‰
    val factorFilter: FactorFilter,
){
    factorFilter: FactorFilter,
) : BaseAnalysisConfig(factorFilter) {
    // è¿žç»­çªå˜æ•°æ®ä¸ªæ•°
    var mutationNum = 2

    // çªå˜çŽ‡
    var mutationRate = .5

    // æ±‚滑动平均值的数据组个数
    var changeTrendGroup = 12

    // æ»‘动平均值连续
    var changeTrendInterval = 12
    var changeTrendRate = .2

    // æ»‘动平均值变化率异常连续次数
    var changeTrendTimes = 3


 src/main/kotlin/com/flightfeather/uav/biz/dataanalysis/model/ExceptionResult.kt

@@ -1,16 +1,20 @@
package com.flightfeather.uav.biz.dataanalysis.model

import com.flightfeather.uav.biz.FactorFilter
import com.flightfeather.uav.biz.dataanalysis.BaseExceptionResult
import com.flightfeather.uav.common.utils.DateUtil
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.socket.eunm.FactorType
import java.math.BigDecimal
import java.util.*

/**
 * å¼‚常结果
 */
open class ExceptionResult {
open class ExceptionResult(): BaseExceptionResult() {
    var missionCode: String? = null
    var deviceCode: String? = null
    var exception: String? = null
@@ -40,7 +44,64 @@
    // ç›¸å…³ä¼ä¸šåç§°ï¼ˆåç§°ä¹‹é—´;分隔）
    var relatedSceneName: List<String>? = null
    var relatedSceneList: List<SceneInfo?>? = null
    // å¼‚常数据，头尾可能包含一定量的偏移
    // å¼‚常数据
    var dataList: MutableList<BaseRealTimeData> = mutableListOf()
    var dataVoList: List<DataVo>? = null

    constructor(
        start: BaseRealTimeData,
        end: BaseRealTimeData?,
        factor: FactorFilter.SelectedFactor,
        exceptionData: List<BaseRealTimeData>,
        missionCode: String?,
        eType: ExceptionType,
    ) : this() {
        this.missionCode = missionCode
        deviceCode = start.deviceCode
        exception = eType.des
        exceptionType = eType.value
        factorId = factor.main.value
        factorName = factor.main.des
        subFactorId = factor.subs.map { it.value }
        subFactorName = factor.subs.map { it.des }
        selectedFactor = factor
        startDate = start.dataTime
        endDate = end?.dataTime
        startTime = DateUtil.instance.dateToString(start.dataTime, DateUtil.DateStyle.HH_MM_SS)
        endTime = DateUtil.instance.dateToString(end?.dataTime, DateUtil.DateStyle.HH_MM_SS) ?: startTime
        startData = start.getByFactorType(factor.main)
        endData = end?.getByFactorType(factor.main) ?: startData

        val avgData = exceptionData.avg()
        // æ±‚取污染数据的中心坐标
        longitude = avgData.longitude
        latitude = avgData.latitude
        // æ±‚取主污染因子的均值和范围
        val s = dataSummary(exceptionData, factor.main)
        avg = s.first
        min = s.second
        max = s.third

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

    private fun dataSummary(exceptionData: List<BaseRealTimeData?>, factorType: FactorType): Triple<Float, Float, Float> {
        var min = -1f
        var max = -1f
        var total = 0f
        var count = 0
        exceptionData.forEach {
            val value = it?.getByFactorType(factorType) ?: return@forEach
            if (min == -1f || min > value) {
                min = value
            }
            if (max == -1f || max < value) {
                max = value
            }
            total += value
            count++
        }
        val avg = if (count == 0) 0f else total / count
        return Triple(avg, min, max)
    }
}

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

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

import com.flightfeather.uav.biz.FactorFilter
import com.flightfeather.uav.biz.dataanalysis.BaseAnalysisConfig

/**
 * å®žæ—¶èµ°èˆªæ±¡æŸ“溯源计算参数
 * @date 2025/5/13
 * @author feiyu02
 */
class RealTimeAnalysisConfig(
    // å› å­ç­›é€‰
    factorFilter: FactorFilter,
) : BaseAnalysisConfig(factorFilter) {

    // é™å®šè·ç¦»å†…（单位：米）
    var distanceLimit = 1000

    // é™å®šæ—¶é—´å†…（单位：分钟）
    var timeLimit = 2

    // çªå˜æ•°æ®ä¸ªæ•°
    var mutationNum = 3

    // çªå˜çŽ‡
    var mutationRate = .2

    // æº¯æºæœ‰æ•ˆæœ€å¤§é£Žé€Ÿï¼Œ5分钟不超过2公里的风速（暂定）
    var sourceTraceWindSpeedLimit = 6.7
}

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

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

import com.flightfeather.uav.biz.FactorFilter
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.domain.entity.BaseRealTimeData
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.socket.handler.UnderwayWebSocketServerHandler
import com.google.gson.Gson

/**
 * å®žæ—¶èµ°èˆªæ±¡æŸ“溯源
 * @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)

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

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

    init {
        initTask(config)
    }

    // è®¡ç®—历史任务
    fun addOneData(data: BaseRealTimeData) {
        taskList

    }

    private fun exceptionCallback(ex: RealTimeExceptionResult) {
        if (sourceTrace(ex, config)) {
            underwayWebSocketServerHandler.broadcast(GsonUtils.gson.toJson(ex))
        }
    }

    private fun sourceTrace(ex: RealTimeExceptionResult, config: RealTimeAnalysisConfig):Boolean {
        val avgData = ex.dataList.avg()
        if (avgData.windSpeed!! > config.sourceTraceWindSpeedLimit) {
            return false
        }

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

//        avgData.longitude
//        avgData.latitude
//        avgData.windDirection
        return false
    }

}

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

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,43 @@
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.dataanalysis.model.ExceptionResult
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 RealTimeExceptionContinuous(config: RealTimeAnalysisConfig) :
    BaseExceptionContinuous<RealTimeAnalysisConfig, RealTimeExceptionResult>(config) {

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

    var callback: NewExceptionCallback? = null

    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/RealTimeExceptionValueMutation.kt

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,92 @@
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
import com.flightfeather.uav.domain.entity.BaseRealTimeData
import com.flightfeather.uav.socket.eunm.FactorType
import java.time.Duration
import java.time.LocalDateTime
import java.time.ZoneId

/**
 * é‡çº§çªå˜å¼‚常溯源
 * @date 2025/5/13
 * @author feiyu02
 */
class RealTimeExceptionValueMutation : RealTimeExceptionContinuous {

    constructor(config: RealTimeAnalysisConfig) : super(config)

    constructor(config: RealTimeAnalysisConfig, callback: NewExceptionCallback) : super(config, callback)

    /**
     * æœ¬å¼‚常的连续发生次数会根据异常的程度变化
     * å½“突变的量级超过设定值1倍时,连续发生次数要求减少1倍
     */
    private var special = false

    override fun getExceptionType(): ExceptionType {
        return ExceptionType.TYPE4
    }

    override fun judgeException(p: BaseRealTimeData?, n: BaseRealTimeData): MutableMap<FactorType, Boolean> {
        val res = mutableMapOf<FactorType, Boolean>()
        config.factorFilter.mainList().forEach { f ->
            if (p?.getByFactorType(f) == null || n.getByFactorType(f) == null) {
                res[f] = (false)
                return@forEach
            }
            val pValue = p.getByFactorType(f)!!
            val nValue = n.getByFactorType(f)!!
            // è®¡ç®—后一个数据相比于前一个数据的变化率
            val r = (nValue - pValue) / pValue
            // å½“变化率为正数（即数据上升时），且大于设定值时，认为是异常情况
            val b1 = r >= (2 * config.mutationRate)
            val b2 = r >= config.mutationRate
            if (b1) special = true
            res[f] = (b1 || b2)
        }

        return res
    }

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

        val b1 = special && (eIndex - sIndex) >= (config.mutationNum / 2)
        val b2 = (eIndex - sIndex) >= config.mutationNum
        special = false
        return b1 || b2
    }

    override fun needCut(tag: Tag): Boolean {
        // æŒ‰ç…§æ—¶é•¿å’Œè·ç¦»é™åˆ¶å°†å¼‚常截取
        if (tag.exceptionData.isEmpty()) return false

        val se = tag.exceptionData.first()
        val ee = tag.exceptionData.last()

        val sTime = LocalDateTime.ofInstant(se.dataTime?.toInstant(), ZoneId.systemDefault())
        val eTime = LocalDateTime.ofInstant(ee.dataTime?.toInstant(), ZoneId.systemDefault())
        val duration = Duration.between(sTime, eTime).toMinutes()
        // æ•°æ®é‡‡æ ·çš„时长超过限制时，需要截取
        val b1 = duration > config.timeLimit

        // èµ°èˆªæ•°æ®çš„距离超过限制时，需要截取
        val b2 = if (se.longitude == null || se.latitude == null || ee.longitude == null || ee.latitude == null) {
            false
        } else {
            val distance = MapUtil.getDistance(
                se.longitude!!.toDouble(), se.latitude!!.toDouble(), ee.longitude!!
                    .toDouble(), ee.latitude!!.toDouble()
            )
            distance > config.distanceLimit
        }

        return b1 || b2
    }
}

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

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

import com.flightfeather.uav.biz.FactorFilter
import com.flightfeather.uav.biz.dataanalysis.BaseExceptionResult
import com.flightfeather.uav.biz.dataanalysis.model.ExceptionType
import com.flightfeather.uav.common.utils.DateUtil
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.socket.eunm.FactorType
import java.math.BigDecimal

/**
 *
 * @date 2025/5/13
 * @author feiyu02
 */
class RealTimeExceptionResult() : BaseExceptionResult() {

    var deviceCode: String? = null

    var exception: String? = null
    var exceptionType: Int? = null

    var factorId: Int? = null
    var factorName: String? = null
    var subFactorId: List<Int>? = null
    var subFactorName: List<String>? = null
    var selectedFactor: FactorFilter.SelectedFactor? = null

    var startTime: String? = null
    var endTime: String? = null

    var startData: Float? = null
    var endData: Float? = null
    var avg: Float? = null
    var min: Float? = null
    var max: Float? = null

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

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

    // æº¯æºä¼ä¸š
    var relatedSceneList: List<SceneInfo?>? = null

    constructor(
        start: BaseRealTimeData,
        end: BaseRealTimeData?,
        factor: FactorFilter.SelectedFactor,
        exceptionData: List<BaseRealTimeData>,
        eType: ExceptionType,
    ) : this() {
        deviceCode = start.deviceCode
        exception = eType.des
        exceptionType = eType.value
        factorId = factor.main.value
        factorName = factor.main.des
        subFactorId = factor.subs.map { it.value }
        subFactorName = factor.subs.map { it.des }
        selectedFactor = factor
        startTime = DateUtil.instance.dateToString(start.dataTime, DateUtil.DateStyle.HH_MM_SS)
        endTime = DateUtil.instance.dateToString(end?.dataTime, DateUtil.DateStyle.HH_MM_SS) ?: startTime
        startData = start.getByFactorType(factor.main)
        endData = end?.getByFactorType(factor.main) ?: startData

        val avgData = exceptionData.avg()
        // æ±‚取污染数据的中心坐标
        longitude = avgData.longitude
        latitude = avgData.latitude
        // æ±‚取主污染因子的均值和范围
        val s = dataSummary(exceptionData, factor.main)
        avg = s.first
        min = s.second
        max = s.third

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

    private fun dataSummary(
        exceptionData: List<BaseRealTimeData?>,
        factorType: FactorType,
    ): Triple<Float, Float, Float> {
        var min = -1f
        var max = -1f
        var total = 0f
        var count = 0
        exceptionData.forEach {
            val value = it?.getByFactorType(factorType) ?: return@forEach
            if (min == -1f || min > value) {
                min = value
            }
            if (max == -1f || max < value) {
                max = value
            }
            total += value
            count++
        }
        val avg = if (count == 0) 0f else total / count
        return Triple(avg, min, max)
    }
}

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

@@ -1,8 +1,9 @@
package com.flightfeather.uav.common.utils

import com.google.gson.Gson
import com.google.gson.GsonBuilder
import com.google.gson.JsonParser
import java.util.ArrayList
import java.time.LocalDateTime

/**
 * @author riku
@@ -11,6 +12,10 @@
 */
object GsonUtils {

    val gson: Gson = GsonBuilder().setDateFormat("yyyy-MM-dd HH:mm:ss")
        .registerTypeAdapter(LocalDateTime::class.java, LocalDateTimeAdapter())
        .create()

    fun getNoteJsonString(jsonString: String, note: String): String {
        if (jsonString.isEmpty()) {
            throw RuntimeException("getNoteJsonString jsonString empty")

 src/main/kotlin/com/flightfeather/uav/common/utils/LocalDateTimeAdapter.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,30 @@
package com.flightfeather.uav.common.utils;

import com.google.gson.*;

import java.lang.reflect.Type;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

/**
 * LocalDateTime类型的时间格式序列化和反序列化类
 * by hc 2024.12.6
 */
public class LocalDateTimeAdapter implements JsonDeserializer<LocalDateTime>, JsonSerializer<LocalDateTime> {
    private final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    @Override
    public JsonElement serialize(LocalDateTime src, Type typeOfSrc, JsonSerializationContext context) {
        return new JsonPrimitive(dateTimeFormatter.format(src));
    }

    @Override
    public LocalDateTime deserialize(JsonElement json, Type typeOfT, JsonDeserializationContext context) throws JsonParseException {
        try {
            return LocalDateTime.parse(json.getAsString(), dateTimeFormatter);
        } catch (Exception e) {
            throw new JsonParseException(e);
        }
    }
}


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

@@ -51,6 +51,7 @@

    /**
     * èŽ·å–两经纬度间的距离
     * @return è¿”回两点间距离，单位：米
     */
    fun getDistance(lng1: Double, lat1: Double, lng2: Double, lat2: Double): Double {
//        lat1 = lat1 || 0;
@@ -167,7 +168,7 @@
     * åˆ¤æ–­åæ ‡ç‚¹æ˜¯å¦åœ¨å¤šè¾¹å½¢å†…部
     */
    fun isPointInPolygon(point: Pair<Double, Double>, polygon: List<Pair<Double, Double>>): Boolean {
        if (polygon.size < 4) throw IllegalArgumentException("not a polygon")
        if (polygon.size < 3) throw IllegalArgumentException("not a polygon")

        // ä¸åœ¨å››è‡³èŒƒå›´å†…，则一定不在多边形内
        if (!inBBox(point, polygon)) return false

 src/main/kotlin/com/flightfeather/uav/domain/entity/BaseRealTimeData.kt

@@ -10,6 +10,8 @@
import java.time.ZoneId
import java.util.*
import javax.persistence.Column
import javax.persistence.GeneratedValue
import javax.persistence.GenerationType
import javax.persistence.Id
import kotlin.math.atan
import kotlin.math.cos
@@ -21,6 +23,7 @@
 */
open class BaseRealTimeData {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    var id: Int? = null

    @Column(name = "device_code")

 src/main/kotlin/com/flightfeather/uav/domain/repository/AirDataRep.kt

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

import com.flightfeather.uav.domain.entity.BaseRealTimeData
import com.flightfeather.uav.domain.entity.RealTimeData
import com.flightfeather.uav.lightshare.bean.DataVo
import com.flightfeather.uav.socket.bean.AirDataPackage
@@ -26,6 +27,6 @@
     */
    fun savePrepData(dataList: List<RealTimeData>): Int

    fun savePrepData2(dataList: List<DataVo>): Int
    fun savePrepData2(dataList: List<DataVo>): List<BaseRealTimeData>

}

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

@@ -127,8 +127,8 @@
        return count
    }

    override fun savePrepData2(dataList: List<DataVo>): Int {
        var count = 0
    override fun savePrepData2(dataList: List<DataVo>): List<BaseRealTimeData> {
        val res = mutableListOf<BaseRealTimeData>()
        dataList.forEach {vo ->
            when (UWDeviceType.getType(vo.deviceCode)) {
                UWDeviceType.VEHICLE -> {
@@ -139,7 +139,7 @@
                    calibration(d, UWDeviceType.VEHICLE)
                    /***************************************************************************************************/
                    realTimeDataVehicleMapper.insert(d)
                    count++
                    res.add(d)
                }
                UWDeviceType.UAV -> {
                    val d = RealTimeDataUav()
@@ -164,7 +164,7 @@
                    }
                    /***************************************************************************************************/
                    realTimeDataUavMapper.insert(d)
                    count++
                    res.add(d)
                }
                UWDeviceType.GRID -> {
                    val d = RealTimeDataGrid()
@@ -196,7 +196,7 @@
//                    d.h2s = d.h2s?.let { sqrt(it) * 2 }
                    /**************************************************************************/
                    realTimeDataGridMapper.insert(d)
                    count++
                    res.add(d)
                }
                UWDeviceType.BOAT -> {

@@ -204,7 +204,7 @@
                else -> Unit
            }
        }
        return count
        return res
    }

    private fun dataTransform(vo: RealTimeData, bean: BaseRealTimeData) {

 src/main/kotlin/com/flightfeather/uav/lightshare/service/impl/RealTimeDataServiceImpl.kt

@@ -395,7 +395,7 @@
            println("当前页数：$page")
            val dataList = res.data ?: emptyList()
            val result = epwDataPrep.mDataPrep2(dataList)
            count += airDataRep.savePrepData2(result)
            count += airDataRep.savePrepData2(result).size
            page++
        }


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

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

import com.flightfeather.uav.socket.handler.ServerHandler
import com.flightfeather.uav.socket.handler.UnderwayWebSocketServerHandler
import com.flightfeather.uav.socket.processor.BaseProcessor
import io.netty.bootstrap.ServerBootstrap
import io.netty.channel.ChannelHandler
@@ -9,15 +11,22 @@
import io.netty.channel.socket.nio.NioServerSocketChannel
import io.netty.channel.socket.nio.NioSocketChannel
import io.netty.handler.codec.LineBasedFrameDecoder
import io.netty.handler.codec.http.HttpObjectAggregator
import io.netty.handler.codec.http.HttpServerCodec
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler
import io.netty.handler.codec.string.StringDecoder
import io.netty.handler.codec.string.StringEncoder
import org.springframework.stereotype.Component
import java.nio.charset.Charset

/*********************************************************************************
 * èµ°èˆªç›‘测数据socket长连接服务端
 * ç”¨äºŽæŽ¥æ”¶è§£æžèµ°èˆªç›‘测数据，前端监测设备目前包括车载走航、无人机走航以及无人船走航三种类型
 * *******************************************************************************/
class UnderwaySocketServer {
@Component
class UnderwaySocketServer(
    private val underwayWebSocketServerHandler:UnderwayWebSocketServerHandler
) {

    private val bossGroup = NioEventLoopGroup()
    private val workerGroup = NioEventLoopGroup()
@@ -28,6 +37,10 @@

    fun startElectricServer(port: Int, processor: BaseProcessor) {
        electricServer(processor)?.bind(port)?.sync()
    }

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

    fun stopServer() {
@@ -75,4 +88,17 @@
                ?.addLast(ServerHandler(processor))
        }
    })

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

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

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

import io.netty.channel.ChannelHandlerContext
import io.netty.channel.ChannelInboundHandlerAdapter
import java.text.SimpleDateFormat
import java.util.*

/**
 * socket消息处理积基类
 * @date 2025/5/13
 * @author feiyu02
 */
abstract class BaseHandler : ChannelInboundHandlerAdapter() {

    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?) {
        println("------【${tag}】IP激活：[ip:${ctx?.channel()?.remoteAddress()}] ${
            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?) {
        println("------【${tag}】端口有IP不活动：[ip:${ctx?.channel()?.remoteAddress()}] ${
            SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(
                Date()
            )}")
        super.channelInactive(ctx)
    }

    @Deprecated("Deprecated in Java")
    override fun exceptionCaught(ctx: ChannelHandlerContext?, cause: Throwable?) {
        cause?.printStackTrace()
        ctx?.close()
    }
}

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

ÎļþÃû´Ó src/main/kotlin/com/flightfeather/uav/socket/ServerHandler.kt ÐÞ¸Ä
@@ -1,10 +1,9 @@
package com.flightfeather.uav.socket
package com.flightfeather.uav.socket.handler

import com.flightfeather.uav.socket.processor.BaseProcessor
import io.netty.channel.ChannelHandlerContext
import io.netty.channel.ChannelInboundHandlerAdapter
import io.netty.util.AttributeKey
import org.ietf.jgss.MessageProp
import java.lang.StringBuilder
import java.text.SimpleDateFormat
import java.util.*

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

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

import io.netty.channel.ChannelHandlerContext
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame
import org.springframework.stereotype.Component

/**
 *
 * @date 2025/5/13
 * @author feiyu02
 */
@Component
class UnderwayWebSocketServerHandler : BaseHandler() {

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

    override var tag: String = "UAV-WS"

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

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

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

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

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

@@ -1,12 +1,20 @@
package com.flightfeather.uav.socket.processor

import com.flightfeather.uav.biz.FactorFilter
import com.flightfeather.uav.biz.sourcetrace.RealTimeExceptionAnalysisController
import com.flightfeather.uav.common.location.LocationRoadNearby
import com.flightfeather.uav.domain.entity.BaseRealTimeData
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.SegmentInfoRep
import com.flightfeather.uav.socket.bean.AirDataPackage
import com.flightfeather.uav.socket.decoder.AirDataDecoder
import com.flightfeather.uav.socket.decoder.DataPackageDecoder
import com.flightfeather.uav.socket.eunm.AirCommandUnit
import com.flightfeather.uav.socket.eunm.FactorType
import com.flightfeather.uav.socket.eunm.UWDeviceType
import com.flightfeather.uav.socket.handler.UnderwayWebSocketServerHandler
import io.netty.channel.ChannelHandlerContext
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Component
@@ -21,16 +29,17 @@
 */

@Component
class UnderwayProcessor : BaseProcessor() {
class UnderwayProcessor(
    private val airDataRep: AirDataRep,
    private val realTimeDataRep: RealTimeDataRep,
    private val locationRoadNearby: LocationRoadNearby,
    private val segmentInfoRep: SegmentInfoRep,
    private val underwayWebSocketServerHandler: UnderwayWebSocketServerHandler,
) : BaseProcessor() {

    companion object {
        private lateinit var instance: UnderwayProcessor

        private const val TAG = "UAV"
    }

    @Autowired
    lateinit var airDataRep: AirDataRep

    private val airDataDecoder = AirDataDecoder.instance
    private val dataPackageDecoder = DataPackageDecoder()
@@ -38,10 +47,24 @@
    // æ•°æ®é¢„处理函数
    private val dataProcessMap = mutableMapOf<String?, EPWDataPrep>()

    @PostConstruct
    fun init() {
        instance = this
    }
    // å®žæ—¶èµ°èˆªæ±¡æŸ“溯源处理器
    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()
        )

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

@@ -53,7 +76,11 @@
            //保存
            deviceSession.saveDevice(packageData.deviceCode, ctx)
            saveToTxt(msg)
            saveToDataBase(packageData)
            saveToDataBase(packageData)?.takeIf { it.isNotEmpty() }?.get(0)?.let {
                // å°†èµ°èˆªæ•°æ®ä¼ å…¥å¼‚常处理器
                realTimeExceptionAnalysisController.addOneData(it)
            }

        } else {
            println("------${TAG}数据BCC校验失败，舍弃 [${SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(Date())}]")
        }
@@ -62,21 +89,23 @@
    /**
     * ä¿å­˜è‡³æ•°æ®åº“
     */
    fun saveToDataBase(dataPackage: AirDataPackage) {
    fun saveToDataBase(dataPackage: AirDataPackage): List<BaseRealTimeData>? {
        when (dataPackage.commandUnit) {
            AirCommandUnit.AirData.value -> {
                // ä»¥json格式存储原始数据
                instance.airDataRep.saveAirData(dataPackage)
                airDataRep.saveAirData(dataPackage)
                // è¿›è¡Œé¢„处理后，存储至对应数据表
                if (!dataProcessMap.containsKey(dataPackage.deviceCode)) {
                    // æ¯å°è®¾å¤‡æœ‰å•ç‹¬çš„数据预处理对象
                    dataProcessMap[dataPackage.deviceCode] = EPWDataPrep(UWDeviceType.getType(dataPackage.deviceCode))
                }
                dataProcessMap[dataPackage.deviceCode]?.run {
                return dataProcessMap[dataPackage.deviceCode]?.run {
                    val list = this.mDataPrep2(dataPackage)// æ•°æ®å¹³æ»‘处理
                    instance.airDataRep.savePrepData2(list)// æŒ‰ç…§è®¾å¤‡ç±»åž‹å­˜å‚¨è‡³å¯¹åº”数据表
                    airDataRep.savePrepData2(list)// æŒ‰ç…§è®¾å¤‡ç±»åž‹å­˜å‚¨è‡³å¯¹åº”数据表
                }
            }

            else -> return emptyList()
        }
    }

@@ -108,7 +137,7 @@
    fun encodeToBytes(msg: String): ByteArray {
        val list = msg.split(" ")
        val bytes = ByteArray(list.size)
        for (i in 0 until list.size) {
        for (i in list.indices) {
            bytes[i] = list[i].toInt(16).toByte()
        }