2025.7.3 1. 新增动态污染溯源新的判定逻辑

feiyu02

2025-07-03 c56e1e74426238939f229f0005828d05089715ff

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

@@ -1,199 +1,211 @@
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.ExceptionTag
import com.flightfeather.uav.domain.entity.BaseRealTimeData
import com.flightfeather.uav.lightshare.eunm.ExceptionStatusType
import com.flightfeather.uav.socket.eunm.FactorType
import java.time.Duration

/**
 * 连续类型的异常分析基类,适用于当前数据与相邻数据之间有关联关系的情况
 */
abstract class BaseExceptionContinuous(config: DataAnalysisConfig) : BaseExceptionAnalysis(config) {
abstract class BaseExceptionContinuous<T : ExceptionTag, V : BaseAnalysisConfig, Y : BaseExceptionResult>(
    config: V, private val tagClz: Class<T>
) : BaseExceptionAnalysis<V, Y>(config) {

    companion object {
        // 记录异常数据段时，分别向起始前和末尾后额外记录的数据个数偏移量
        private const val OFFSET = 10
    }

    inner class Tag {
        // 起始数据下标
        var sIndex = 0

        // 起始数据对象
        var startData :BaseRealTimeData? = null

        // 末尾数据下标
        var eIndex = -1

        // 末尾数据对象
        var endData: BaseRealTimeData? = null

        // 异常数据段
        var exceptionData = mutableListOf<BaseRealTimeData>()

        // 是否存在异常
        var existException = false

        fun refreshAfterCheckResult(data: BaseRealTimeData) {
            sIndex = eIndex
            startData = data
            exceptionData.clear()
            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>()
    protected val tagMap = mutableMapOf<FactorType, T>()

    // 起始数据与末尾数据间隔
    open var durationCount = 1

    // 末尾数据对象
    protected var lastData: BaseRealTimeData? = null

    /**
     * 判断是否满足异常条件
     * 后置判断：当相邻数据时间不连续时，或者满足自定义条件时，对之前已有的异常进行记录
     */
    open fun afterExcCheck(isContinue: Boolean, tag: T, hasException: Boolean?): Boolean {
        return !isContinue || needCut(tag, hasException)
    }

    /**
     * 立即判断：当出现异常时，缓存异常数据的同时，立即对已有异常进行判断是否满足异常结果要求
     */
    open fun immeExcCheck(tag: T, factorType: FactorType): Boolean {
        return false
    }

    /**
     * 判断相邻数据是否连续
     */
    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 judgeDataScale(p: BaseRealTimeData?, n: BaseRealTimeData): MutableMap<FactorType, Boolean> {
        val res = mutableMapOf<FactorType, Boolean>()
        config.factorFilter.mainList().forEach { f -> res[f] = true }
        return res
    }

    /**
     * 判断前后数据是否满足异常条件
     */
    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: T, factorType: FactorType?): Boolean

    /**
     * 判断监测因子是否出现异常
     */
    open fun judge(p: BaseRealTimeData?, n: BaseRealTimeData): MutableMap<FactorType, Boolean> {
        val jds = judgeDataScale(p, n)
        val jex = judgeException(p, n)
        val res = mutableMapOf<FactorType, Boolean>()
        jds.forEach { (t, u) ->
            res[t] = u && jex[t] ?: false
        }
        return res
    }

    /**
     * 异常数据的截取判断
     * @return
     */
    open fun needCut(tag: T, hasException: Boolean?): Boolean {
        // 默认判断条件为 当异常不再重复出现时，形成异常结果
        return tag.exceptionExisted && hasException == 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()
        config.factorFilter.mainList().forEach { f ->
            tagMap[f] = tagClz.newInstance()
        }
    }

    override fun onNextData(data: BaseRealTimeData) {
        val isContinue = isContinuous(lastData, data)
        val hasException = judgeException(lastData, data)
        config.factorFilter.selectedList.forEach {s->
        val hasException = judge(lastData, data)
        config.factorFilter.selectedList.forEach { s ->
            val f = s.main
            tagMap[f]?.let {
                it.addHistoryData(data)

                it.eIndex++
                // 起始数据
                it.endData = lastData
                if (it.endData == null) {
                    it.refreshAfterCheckResult(data)
                it.endData = data
                if (it.startData == null) {
                    it.refreshWithNextException(data)
                }
                // 判断相邻数据是否连续并且是否满足异常判断
                if (!isContinue) {
                    checkResult(s)
                    // 数据不连续时，记录异常情况
                    if (it.eIndex - it.sIndex >= durationCount) {
                        it.refreshAfterCheckResult(data)

                // 对于异常的生成分别执行后置判断、和立即判断
                // 1. 后置判断：当相邻数据时间不连续时，或者满足自定义条件时，对之前已有的异常进行记录，形成异常结果
                if (afterExcCheck(isContinue, it, hasException[f])) {
                    // 数据不连续时或者满足主动截断条件时，记录异常情况
                    recordException(s, it, data)
                }
                // 2. 立即判断：当出现异常时，缓存异常数据的同时，立即对已有异常进行判断是否满足异常结果要求
                else if (hasException[f] == true) {
                    // 有异常出现时，记录异常数据
                    it.addExceptionData(data)
                    // 当立即判断通过时，形成异常结果
                    if (immeExcCheck(it, f)) {
                        recordException(s, it, 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)
                        }
                    }
                }
                // 3. 数据正常，无任何异常时d
                // TODO("2025.6.3：其他子类的此处刷新逻辑待完成“)
                else {
                    it.refreshWithNextException(data)
                }
            }
        }
//        repeat(config.factorCount) { i ->
//            eIndex[i]++
//            // 起始数据
//            if (lastData == null) {
//                refreshAfterCheckResult(i, data)
//            }
//            // 判断相邻数据是否连续并且是否满足异常判断
//            if (!isContinue) {
//                checkResult()
//                // 数据不连续时，记录异常情况
//                if (eIndex[i] - sIndex[i] >= durationCount) {
//                    refreshAfterCheckResult(i, data)
//                }
//            } else {
//                if (hasException[i]) {
//                    existException[i] = true
//                    exceptionData[i].add(data)
//                } else {
//                    // 异常不再重复出现时，记录异常情况
//                    checkResult()
//                    if (eIndex[i] - sIndex[i] >= durationCount) {
//                        refreshAfterCheckResult(i, data)
//                    }
//                }
//            }
//        }
        lastData = data
    }

    override fun onDone() {
        checkResult()
        checkResult(exceptionStatus = ExceptionStatusType.Ended)
    }

//    fun refreshAfterCheckResult(i:Int, data: BaseRealTimeData) {
//        sIndex[i] = eIndex[i]
//        startData[i] = data
//        exceptionData[i].clear()
//        exceptionData[i].add(data)
//    }
    /**
     * 异常结束，记录异常
     * 判断已有的异常数据是否满足异常条件，满足则记录，不满足则略过
     */
    fun recordException(factor: FactorFilter.SelectedFactor, tag: T, data: BaseRealTimeData) {
        checkResult(factor, ExceptionStatusType.Ended)
        tag.refreshWithNextException(data)
    }

    /**
     * 检查连续异常结束时，是否符合异常存储条件
     */
    open fun checkResult(factor: FactorFilter.SelectedFactor? = null) {
    open fun checkResult(
        factor: FactorFilter.SelectedFactor? = null,
        exceptionStatus: ExceptionStatusType = ExceptionStatusType.InProgress
    ) {
        val tag = tagMap[factor?.main]
        if (factor != null && tag != null) {
            if (tag.existException && judgeDuration(tag.sIndex, tag.eIndex - 1)) {
                tag.startData?.let {
                    resultList.add(newResult(it, lastData, factor, tag.exceptionData))
                }
                tag.existException = false
            if (tag.exceptionExisted && judgeExceptionCount(tag, factor.main)) {
                onNewException(tag, factor, exceptionStatus)
            }
        } 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
                if (tag1.exceptionExisted && judgeExceptionCount(tag1, f.main)) {
                    onNewException(tag1, f, exceptionStatus)
                }
            }
//            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
//                }
//            }
        }
    }

    /**
     * 新增或更新一条异常
     */
    open fun onNewException(tag: T, factor: FactorFilter.SelectedFactor, exceptionStatus: ExceptionStatusType) {
        if (tag.startData == null) return
//        val ex = newResult(tag.startData!!, tag.endData, factor, tag.exceptionData)
        val ex = newResult(tag, factor)
            .apply { status = exceptionStatus.value }
        // 异常已创建时，更新异常信息
        if (tag.exceptionCreated) {
            // 将最新的异常的guid赋值给ex
            val lastEx = tag.exceptionResult.last()
            ex.guid = lastEx.guid
            tag.exceptionResult.removeLast()
            tag.exceptionResult.add(ex)
        }
        // 异常未创建时，新建异常信息
        else {
            tag.exceptionResult.add(ex)
//            resultList.add(ex)
            tag.exceptionCreated = true
        }
    }

    /**
     * 生成一条异常分析结果
     */
    abstract fun newResult(tag:T, factor: FactorFilter.SelectedFactor): Y

}