2025.7.21 1. 修改动态溯源异常判断逻辑

feiyu02

2025-07-21 38d72198bfcced01ed9513b978163e5cd1d84625

2025.7.21
1. 修改动态溯源异常判断逻辑

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

@@ -15,6 +15,11 @@
    config: V, private val tagClz: Class<T>,
) : BaseExceptionAnalysis<V, Y>(config) {

    enum class JudgeMethod(val des: String) {
        M1("在一定的空间和时间范围内，数据累计出现N次异常后，认为该异常成立"),
        M2("要求数据不间断连续出现N次异常后，认为该异常成立"),
    }

    companion object {
        // è®°å½•å¼‚常数据段时，分别向起始前和末尾后额外记录的数据个数偏移量
        private const val OFFSET = 10
@@ -40,6 +45,13 @@
     * å¼‚常结果
     */
    protected val result = mutableListOf<Y>()

    /**
     * ä¸é€‚用于此异常类型的监测因子
     */
    open var excludedFactor: List<FactorType> = emptyList()

    abstract var judgeMethod: JudgeMethod

    /**
     * ç«‹å³åˆ¤æ–­ï¼šå½“出现异常时，缓存异常数据的同时，立即对已有异常进行判断是否满足异常结果要求
@@ -98,7 +110,7 @@
     * @return
     */
    open fun needCut(tag: T, hasException: Boolean?, data: BaseRealTimeData): Boolean {
        // é»˜è®¤åˆ¤æ–­æ¡ä»¶ä¸º å½“异常不再重复出现时，形成异常结果
        // é»˜è®¤åˆ¤æ–­æ¡ä»¶ä¸º å½“异常不再重复出现时
        return tag.exceptionExisted && hasException == false
    }

@@ -116,9 +128,10 @@
        val hasException = judge(lastData, data)
        config.factorFilter.selectedList.forEach { s ->
            val f = s.main
            tagMap[f]?.let {
                it.addHistoryData(data)
            // æŽ’除此异常类型不适用的监测因子
            if (excludedFactor.contains(f)) return@forEach

            tagMap[f]?.let {
                it.eIndex++
                // èµ·å§‹æ•°æ®
                it.endData = data
@@ -126,31 +139,13 @@
                    it.refreshWithNextException(data)
                }

                // å¯¹äºŽå¼‚常的生成分别执行后置判断、和立即判断
                // 1. åŽç½®åˆ¤æ–­ï¼šå½“相邻数据时间不连续时，或者满足自定义条件时，对之前已有的异常进行记录，形成异常结果
//                if (afterExcCheck(isContinue, it, hasException[f])) {
//                    // æ•°æ®ä¸è¿žç»­æ—¶æˆ–者满足主动截断条件时，记录异常情况
//                    recordException(s, it, data)
//                }
                // 2. ç«‹å³åˆ¤æ–­ï¼šå½“出现异常时，缓存异常数据的同时，立即对已有异常进行判断是否满足异常结果要求
                if (hasException[f] == true) {
                // æŒ‰ç…§ä¸åŒçš„方式进行异常判断
                when (judgeMethod) {
                    JudgeMethod.M1 -> judgeMethod1(hasException, f, it, data, s)
                    JudgeMethod.M2 -> judgeMethod2(isContinue, hasException, f, it, data, s)
                }

//                    afterExcCheck(isContinue, it, hasException[f])
                    if (needCut(it, hasException[f], data)) {
                        it.refreshWithNextException(data)
                    }
                    // æœ‰å¼‚常出现时，记录异常数据
                    it.addExceptionData(data)
                    // å½“立即判断通过时，形成异常结果
                    if (immeExcCheck(it, f)) {
                        recordException(s, it, data)
                    }
                }
                // 3. æ•°æ®æ­£å¸¸ï¼Œæ— ä»»ä½•å¼‚常时d
                // TODO("2025.6.3：其他子类的此处刷新逻辑待完成“)
//                else {
//                    it.refreshWithNextException(data)
//                }
                it.addHistoryData(data)
            }
        }
        lastData = data
@@ -164,6 +159,67 @@
    }

    /**
     * æ•°æ®å¼‚常判断方式一
     * åœ¨ä¸€å®šçš„空间和时间范围内，数据累计出现N次异常后，认为该异常成立
     */
    private fun judgeMethod1(
        hasException: MutableMap<FactorType, Boolean>,
        f: FactorType,
        tag: T,
        data: BaseRealTimeData,
        s: FactorFilter.SelectedFactor,
    ) {
        // å‡ºçŽ°å¼‚常
        if (hasException[f] == true) {
            // åˆ¤æ–­æ•°æ®åœ¨ç©ºé—´å’Œæ—¶é—´å˜åŒ–上是否超出限定范围，若超出则删除遗留的异常记录，刷新起始点数据
            if (needCut(tag, hasException[f], data)) {
                tag.refreshWithNextException(data)
            }
            // è®°å½•å¼‚常数据
            tag.addExceptionData(data)
            // å½“立即判断通过时，形成异常结果
            if (immeExcCheck(tag, f)) {
                recordException(s, tag, data)
            }
        }
        // æ•°æ®æ­£å¸¸ï¼Œå¹¶ä¸”没有历史异常数据时，刷新起始点数据
        else if (!tag.exceptionExisted) {
            tag.refreshWithNextException(data)
        }
    }

    /**
     * æ•°æ®å¼‚常判断方式二
     * è¦æ±‚数据不间断连续出现N次异常后，认为该异常成立
     */
    private fun judgeMethod2(
        isContinue: Boolean,
        hasException: MutableMap<FactorType, Boolean>,
        f: FactorType,
        tag: T,
        data: BaseRealTimeData,
        s: FactorFilter.SelectedFactor,
    ) {
        // å½“相邻数据时间不连续时，刷新起始点数据，移除历史异常记录
        if (!isContinue) {
            tag.refreshWithNextException(data)
        }
        // å‡ºçŽ°å¼‚常
        else if (hasException[f] == true) {
            // æœ‰å¼‚常出现时，记录异常数据
            tag.addExceptionData(data)
            // å½“立即判断通过时，形成异常结果
            if (immeExcCheck(tag, f)) {
                recordException(s, tag, data)
            }
        }
        // æ•°æ®æ­£å¸¸ï¼Œåˆ·æ–°èµ·å§‹ç‚¹æ•°æ®ï¼Œç§»é™¤åŽ†å²å¼‚常记录
        else {
            tag.refreshWithNextException(data)
        }
    }

    /**
     * å¼‚常结束，记录异常
     * åˆ¤æ–­å·²æœ‰çš„异常数据是否满足异常条件，满足则记录，不满足则略过
     */

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

@@ -16,6 +16,8 @@
class ExceptionDataExceed(config: DataAnalysisConfig) :
    BaseExceptionContinuousSingle<ExceptionTag, DataAnalysisConfig, ExceptionResult>(config, ExceptionTag::class.java) {

    override var judgeMethod: JudgeMethod = JudgeMethod.M2

    override fun getExceptionType(): ExceptionType = ExceptionType.TYPE2

    override fun judgeException(p: BaseRealTimeData?, n: BaseRealTimeData): MutableMap<FactorType, Boolean> {

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

@@ -20,6 +20,8 @@

    override fun getExceptionType(): ExceptionType = ExceptionType.TYPE4

    override var judgeMethod: JudgeMethod = JudgeMethod.M2

    override fun judgeException(p: BaseRealTimeData?, n: BaseRealTimeData): MutableMap<FactorType, Boolean> {
        val res = mutableMapOf<FactorType, Boolean>()
        config.factorFilter.mainList().forEach { f ->

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

@@ -9,6 +9,9 @@
 * @author feiyu02
 */
open class ExceptionTag {
    companion object {
        const val MAX_HISTORY = 10
    }
    // èµ·å§‹æ•°æ®ä¸‹æ ‡
    var sIndex = 0

@@ -37,8 +40,16 @@

    fun addHistoryData(data: BaseRealTimeData) {
        historyData.add(data)
        if (historyData.size > 15) {
            historyData.removeAt(0)
        if (exceptionData.isNotEmpty()) {
            // ä¿è¯åŽ†å²æ•°æ®åŒ…含所有异常数据（异常数据可能不连续），并且在首个异常数据之前最多再保存10个数据
            val i = historyData.indexOf(exceptionData.first())
            if (i > MAX_HISTORY) {
                historyData = historyData.subList(i - MAX_HISTORY, historyData.size)
            }
        } else {
            if (historyData.size > MAX_HISTORY) {
                historyData = historyData.subList(historyData.size - MAX_HISTORY, historyData.size)
            }
        }
    }


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

@@ -43,7 +43,7 @@
                    .withMain(FactorType.O3)
                    .withMain(FactorType.PM25)
                    .withMain(FactorType.PM10)
                    .withMain(FactorType.VOC)
//                    .withMain(FactorType.VOC)
                    .withCombination(
                        listOf(
                            listOf(FactorType.PM25, FactorType.PM10),
@@ -63,6 +63,8 @@
    private val sceneInfoRep: SceneInfoRep
    private val sourceTraceRep: SourceTraceRep
    private val config: RTExcWindLevelConfig
    private val timer = Timer()
    private var timerTask: TimerTask? = null

    private val taskList = mutableListOf<BaseExceptionAnalysis<RTExcWindLevelConfig, PollutedClue>>()

@@ -92,23 +94,27 @@
     * è®¡ç®—新的一条实时走航数据
     */
    fun addOneData(data: BaseRealTimeData) {
//        println("====================>")
        // è®¡ç®—异常
        taskList.forEach { it.onNextData(data) }
        pollutedSummary.refreshLatestMonitorData(data)
        // é™å®šæ—¶é—´å†…没有新数据传入，则结束当前的计算
        dealOnTimeout()
    }

    /**
     * è¶…时处理，较长时间没有新数据进入，进行初始化操作
     */
    private fun dealOnTimeout() {
        val timer = Timer(true)
        timer.schedule(object : TimerTask() {
//        val timer = Timer()
        timerTask?.cancel()
        timer.purge()
        timerTask = object : TimerTask() {
            override fun run() {
                TODO("Not yet implemented")
                initTask()
            }
        }, 60 * 1000)
        timer.cancel()
        }
        timer.schedule(timerTask, 2 * 60 * 60 * 1000)
    }

    // æ•°æ®çªå˜å¼‚常回调

 src/main/kotlin/com/flightfeather/uav/biz/sourcetrace/config/RTExcWindLevelConfig.kt

@@ -27,7 +27,7 @@
    // æº¯æºæ‰©æ•£åç§»è§’度（单位：度）
    var sourceTraceDegOffset = 120.0
    // å®šæ—¶çº¿ç´¢åˆ†æžæ—¶é—´é—´éš”(单位：分钟)
    var analysisPeriod = 15
    var analysisPeriod = 5
    // å®šæ—¶åˆ†æžé—´éš”中，立即进行线索分析的最小线索量(单位：个)
    var analysisCount = 4


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

@@ -30,6 +30,8 @@

    abstract var changeRate: MutableMap<FactorType, RTExcWindLevelConfig.WindLevelCondition>

    override var judgeMethod: JudgeMethod = JudgeMethod.M2

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

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

@@ -29,10 +29,14 @@
        this.callback = callback
    }

    override var excludedFactor: List<FactorType> = listOf(FactorType.NO2)

    private var callback: NewPolluteClueCallback? = null

    abstract var windLevelCondition: RTExcWindLevelConfig.WindLevelCondition

    override var judgeMethod: JudgeMethod = JudgeMethod.M1

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

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

@@ -32,6 +32,8 @@
        return ExceptionType.TYPE4
    }

    override var judgeMethod: JudgeMethod = JudgeMethod.M2

    override fun judgeException(p: BaseRealTimeData?, n: BaseRealTimeData): MutableMap<FactorType, Boolean> {
        val res = mutableMapOf<FactorType, Boolean>()
        config.factorFilter.mainList().forEach { f ->
@@ -67,7 +69,7 @@
        if (tag.exceptionData.isEmpty()) return false

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

        val sTime = LocalDateTime.ofInstant(se.dataTime?.toInstant(), ZoneId.systemDefault())
        val eTime = LocalDateTime.ofInstant(ee.dataTime?.toInstant(), ZoneId.systemDefault())

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

@@ -26,6 +26,8 @@
        windLevelCondition: RTExcWindLevelConfig.WindLevelCondition?,
    ) : this() {
        distanceType = windLevelCondition?.distanceType
        distanceRange = distanceType?.disRange
        distanceDes = distanceType?.des
        windLevelCondition?.let { sourceTrace(historyData, exceptionData, config, it) }
    }

@@ -40,6 +42,9 @@
    // æ±¡æŸ“可能的发生距离
    var distanceType: DistanceType? = null

    var distanceRange: Pair<Double, Double>? = null
    var distanceDes: String? = null

    /**
     * åå‘溯源
     */

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

@@ -64,6 +64,8 @@

    // æ–°å¢žä¸€æ¡æ±¡æŸ“线索
    fun addClue(pollutedClue: PollutedClue) {
        // å½“溯源未找到风险源时，此次溯源信息不作为线索统计项
        if (pollutedClue.pollutedSource?.sceneList?.isNotEmpty() == true)
        clueList.add(pollutedClue)
//        realTimeSummary()
        analysisOnClueCount()
@@ -71,8 +73,7 @@

    // æ–°å¢žä¸€æ¡æ±¡æŸ“线索
    fun addClueList(pollutedClues: List<PollutedClue>) {
        clueList.addAll(pollutedClues)
        analysisOnClueCount()
        pollutedClues.forEach { addClue(it) }
    }

    // åˆ·æ–°å½“前最新的走航监测数据
@@ -181,6 +182,7 @@

                // å»ºè®®çš„走航路线
                result.direction = AMapService.directionDriving(origin, destination)
                Thread.sleep(200)
            }
            // çº¿ç´¢åˆ†æžå®ŒæˆåŽï¼Œç§»åŠ¨è‡³åŽ†å²çº¿ç´¢åˆ—表
            historyClueList.addAll(clueList)

 src/test/kotlin/com/flightfeather/uav/Test.kt

@@ -168,4 +168,37 @@
    fun foo18() {
        println(-4.382398 in 4.0..Double.MAX_VALUE)
    }

    @Test
    fun foo19() {
        val timer = Timer(true)
//        var running = true
        val task = object : TimerTask() {
            override fun run() {
                println("task run")
                println(Date())
//                running = false
            }
        }
        println(Date())
        timer.schedule(task, 5000)
        task.cancel()
        timer.purge()

        val task2 = object : TimerTask() {
            override fun run() {
                println("task2 run")
                println(Date())
//                running = false
            }
        }
        timer.schedule(task2, 4000)
//        while (running) {
//
//        }
        val sc = Scanner(System.`in`)
        while (sc.hasNext()) {
            println(sc.nextLine())
        }
    }
}

 src/test/kotlin/com/flightfeather/uav/biz/sourcetrace/SourceTraceControllerTest.kt

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

import com.flightfeather.uav.common.utils.DateUtil
import com.flightfeather.uav.domain.entity.Mission
import com.flightfeather.uav.domain.entity.RealTimeDataVehicle
import com.flightfeather.uav.domain.mapper.MissionMapper
import com.flightfeather.uav.domain.repository.MissionRep
import com.flightfeather.uav.domain.repository.SceneInfoRep
import com.flightfeather.uav.domain.repository.SourceTraceRep
import com.flightfeather.uav.lightshare.service.RealTimeDataService
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
import tk.mybatis.mapper.entity.Example

@RunWith(SpringRunner::class)
@SpringBootTest
class SourceTraceControllerTest {

    @Autowired
    lateinit var sceneInfoRep: SceneInfoRep

    @Autowired
    lateinit var sourceTraceRep: SourceTraceRep

    @Autowired
    lateinit var missionMapper: MissionMapper

    @Autowired
    lateinit var realTimeDataService: RealTimeDataService

    @Test
    fun autoSourceTrace() {
        val sourceTraceController = SourceTraceController(sceneInfoRep, sourceTraceRep)
        val missions = missionMapper.selectByExample(Example(Mission::class.java).apply {
            createCriteria().andEqualTo("deviceType", "0a")
                .andLessThanOrEqualTo("startTime", "2024-11-07 15:00:00")
            orderBy("startTime").desc()
        })
        missions.forEach { m ->
            val rtData = realTimeDataService.getSecondData(
                m?.deviceType,
                m?.deviceCode,
                DateUtil.instance.dateToString(m?.startTime, DateUtil.DateStyle.YYYY_MM_DD_HH_MM_SS),
                DateUtil.instance.dateToString(m?.endTime, DateUtil.DateStyle.YYYY_MM_DD_HH_MM_SS),
                null,
                1,
                10000
            )
            rtData.data?.forEach { d ->
                val rtdVehicle = d.toBaseRealTimeData(RealTimeDataVehicle::class.java)
//            Thread.sleep(500)
                sourceTraceController.addOneData(rtdVehicle)
            }
            sourceTraceController.initTask()
        }
//        val rtData = realTimeDataService.getSecondData(
//            "0a",
//            "0a0000000001",
//            "2025-01-06 13:32:00",
//            "2025-01-06 15:52:36",
//            null,
//            1,
//            10000
//        )
//        rtData.data?.forEach { d ->
//            val rtdVehicle = d.toBaseRealTimeData(RealTimeDataVehicle::class.java)
////            Thread.sleep(500)
//            sourceTraceController.addOneData(rtdVehicle)
//        }

    }

}