feiyu02
2025-07-21 38d72198bfcced01ed9513b978163e5cd1d84625
src/main/kotlin/com/flightfeather/uav/biz/sourcetrace/SourceTraceController.kt
@@ -32,7 +32,7 @@
        this.sceneInfoRep = sceneInfoRep
        this.sourceTraceRep = sourceTraceRep
        this.config = if (factorFilter != null) {
            RTExcWindLevelConfig(factorFilter, emptyList())
            RTExcWindLevelConfig(factorFilter)
        } else {
            RTExcWindLevelConfig(
                FactorFilter.builder()
@@ -43,12 +43,14 @@
                    .withMain(FactorType.O3)
                    .withMain(FactorType.PM25)
                    .withMain(FactorType.PM10)
                    .withMain(FactorType.VOC)
                    .create(),
                listOf(
                    listOf(FactorType.PM25, FactorType.PM10),
                    listOf(FactorType.VOC, FactorType.CO),
                )
//                    .withMain(FactorType.VOC)
                    .withCombination(
                        listOf(
                            listOf(FactorType.PM25, FactorType.PM10),
                            listOf(FactorType.VOC, FactorType.CO),
                        )
                    )
                    .create()
            )
        }
        pollutedSummary = PollutedSummary(config) { summaryCallback(it) }
@@ -61,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>>()
@@ -90,52 +94,52 @@
     * 计算新的一条实时走航数据
     */
    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)
    }
    // 数据突变异常回调
    private fun exceptionCallback(ex: List<PollutedClue>) {
        ex.forEach {
            // 溯源污染源信息
            it.searchScenes(sceneInfoRep)
            it.msgType = MsgType.PolClue.value
        }
    private fun exceptionCallback(ex: PollutedClue) {
        // 溯源污染源信息
        ex.searchScenes(sceneInfoRep)
        ex.msgType = MsgType.PolClue.value
        // 广播污染溯源异常结果
        UnderwayWebSocketSender.broadcast(MsgType.PolClue.value, ex)
        sourceTraceRep.insertList(MsgType.PolClue, ex)
        sourceTraceRep.insert(MsgType.PolClue, ex)
        // 记录污染线索
        pollutedSummary.addClueList(ex)
        pollutedSummary.addClue(ex)
    }
    // 数据变化提醒回调
    private fun dataChangeCallback(ex: List<PollutedClue>) {
        ex.forEach {
            // 溯源污染源信息
            it.searchScenes(sceneInfoRep)
            it.msgType = MsgType.DataChange.value
        }
    private fun dataChangeCallback(ex: PollutedClue) {
        // 溯源污染源信息
        ex.searchScenes(sceneInfoRep)
        ex.msgType = MsgType.DataChange.value
        // 广播数据变化提醒
        UnderwayWebSocketSender.broadcast(MsgType.DataChange.value, ex)
        sourceTraceRep.insertList(MsgType.DataChange, ex)
        sourceTraceRep.insert(MsgType.DataChange, ex)
    }
    private fun summaryCallback(ex: AnalysisResult) {