2026.4.10 1. 修复采样时间超过当前时间的数据依旧能存入数据库的bug

riku

6 天以前 20b8d870efbbb89564b599561fc69202ba41223f

 src/main/kotlin/com/flightfeather/uav/model/epw/EPWDataPrep.kt

@@ -1,45 +1,78 @@
package com.flightfeather.uav.model.epw

import com.flightfeather.uav.lightshare.bean.CompanySOP
import com.flightfeather.uav.common.utils.DateUtil
import com.flightfeather.uav.lightshare.bean.DataVo
import com.flightfeather.uav.model.BaseDataPrep
import com.flightfeather.uav.model.BaseSOP
import com.flightfeather.uav.socket.bean.AirData
import com.flightfeather.uav.socket.bean.AirDataPackage
import com.flightfeather.uav.socket.eunm.FactorType
import com.flightfeather.uav.socket.eunm.UWDeviceType
import kotlin.math.max
import kotlin.math.min
import kotlin.math.round
import kotlin.math.sqrt
import kotlin.time.times

class EPWDataPrep : BaseDataPrep<DataVo, CompanySOP>() {
/**
 * 数据平滑预处理
 * 对于最新的一组数据，根据其之前连续的若干数据，进行数据平滑处理
 * @Date 2024.5.16
 */
class EPWDataPrep(val deviceType: UWDeviceType? = UWDeviceType.GRID) : BaseDataPrep<DataVo, BaseSOP>() {

    // 向前检索的数据记录数
    private val ncal = 15

    // 标准差倍数参数
    private val nstd = 3

    // 均值倍数参数
    private val xratio = 3
    // 需要处理的因子类型
    private val calTypes =
//        emptyList<String>()
        WeightType.prep

    // 需要平滑处理的因子类型
    private var calTypes = when (deviceType) {
        UWDeviceType.VEHICLE,
        UWDeviceType.UAV,
        UWDeviceType.BOAT,
        -> WeightType.prepUnderWay
        UWDeviceType.GRID -> WeightType.prepFixed
        else -> WeightType.prepFixed
    }

    // 只需要检查范围和变化幅度的因子类型
    private var rangeTypes = listOf(
        FactorType.PM25.des,
        FactorType.PM10.des,
        FactorType.VOC.des
    )

    // 无需修正的因子类型
    private var noCalTypes = listOf(
        FactorType.TEMPERATURE.des,
        FactorType.HUMIDITY.des,
        FactorType.LNG.des,
        FactorType.LAT.des,
        FactorType.VELOCITY.des,
        FactorType.TIME.des,
        FactorType.WIND_DIRECTION.des,
        FactorType.HEIGHT.des
    )

    private val lastData = mutableListOf<DataVo>()

    override fun mDataPrep(mDataList: List<DataVo>): List<DataVo> {
        mDataList.forEach {
            it.values?.forEach v@{a ->
            it.values?.forEach v@{ a ->
                if (!calTypes.contains(a.factorName)) return@v

                val range = FactorType.getRange(a.factorName) ?: return@v
                // 判断数据是否在合理范围内
                if (a.factorData ?: 0.0 < range.first || a.factorData ?: 0.0 > range.second) {
                if (a.factorData != null && (a.factorData!! < range.first || a.factorData!! > range.second)) {
                    a.factorData = null
                }
            }
        }

//        val newDataList = mutableListOf<DataVo>()
//        mDataList.forEach {
//            newDataList.add(it.copy())
//        }

        var i = ncal
        if (lastData.isNotEmpty()) {
@@ -50,12 +83,12 @@
                val it = mDataList[i].values?.get(y) ?: continue

                if (!calTypes.contains(it.factorName)) continue
                val vMax = FactorType.getVMax(it.factorName) ?: continue
                val vMax = FactorType.getVMin(it.factorName) ?: continue
                it.factorData ?: continue

                if (it.factorData!! > vMax) {
                    val lastDataIndex = i
                    val thisIndex = if (i-ncal<0) 0 else i - ncal
                    val thisIndex = if (i - ncal < 0) 0 else i - ncal
                    val list = mutableListOf<DataVo>()
                    if (lastDataIndex < lastData.size) {
                        list.addAll(lastData.subList(lastDataIndex, lastData.lastIndex + 1))
@@ -63,7 +96,7 @@
                    list.addAll(mDataList.subList(thisIndex, i))

                    // 去除无效值的平均
                    val avg = average(list, it.factorName)
                    val avg = average(list, it.factorName) ?: continue
                    // 去除无效值的标准差
                    val std = standardDeviation(avg.first, list, it.factorName)
                    // 合理最大值
@@ -94,8 +127,127 @@
        return mDataList
    }

    override fun sopPrep(sopList: List<CompanySOP>): List<CompanySOP> {
    override fun sopPrep(sopList: List<BaseSOP>): List<BaseSOP> {
        return sopList
    }


    /**
     * 实时数据平滑处理
     */
    fun mDataPrep2(dataPackage: AirDataPackage): List<DataVo> {
        val vo = dataPackage.toDataVo()
        return mDataPrep2(vo)
    }

    /**
     * 实时数据平滑处理
     */
    fun mDataPrep2(dataVo: DataVo): List<DataVo> {
        return mDataPrep2(listOf(dataVo))
    }

    /**
     * 实时数据平滑处理
     */
    fun mDataPrep2(mDataList: List<DataVo>): List<DataVo> {
        var i = ncal
        if (lastData.isNotEmpty()) {
            i = 0
        }
        while (i < mDataList.size) {
            // 针对每个监测因子，分别做数据平滑处理
            for (y in mDataList[i].values?.indices ?: 0..0) {
                val it = mDataList[i].values?.get(y) ?: continue
                val vMax = FactorType.getVMin(it.factorName) ?: continue
                it.factorData ?: continue

                if (it.factorData!! > vMax) {
                    val lastDataIndex = i
                    val thisIndex = if (i - ncal < 0) 0 else i - ncal
                    val list = mutableListOf<DataVo>()
                    if (lastDataIndex < lastData.size) {
                        list.addAll(lastData.subList(lastDataIndex, lastData.lastIndex + 1))
                    }
                    list.addAll(mDataList.subList(thisIndex, i))

                    // 去除无效值的平均，当所有数据都是无效值时，暂不做处理
                    average(list, it.factorName)?.let { avg ->
                        // 去除无效值的标准差
                        val std = standardDeviation(avg.first, list, it.factorName)
                        // 合理最大值
                        var maxValue = max(avg.first + std * nstd, avg.first + avg.first * xratio)
                        maxValue = max(maxValue, FactorType.getRange(it.factorName)?.second ?: .0)
                        // 合理最小值
                        val minValue = min(avg.first - std * nstd, avg.first / (1 + xratio))

                        // 判断监测因子是否需要进行平滑处理，
                        if (calTypes.contains(it.factorName)) {
                            // 数据不处于合理范围并且有效个数达标时，采用计算所得均值代替原始值
                            if (avg.second > max(ncal / 5, 2)
                                && (it.factorData!! < minValue || it.factorData!! > maxValue)
                            ) {
                                it.factorData = avg.first
                            }
                        }
                        // 判断量级是否在合理范围内以及变化倍率是否在合理范围内
                        else if (rangeTypes.contains(it.factorName)) {
                            if (isInRange(it) != true || excessiveChange(it) == true) {
                                // 采用计算所得均值代替原始值
                                it.factorData = avg.first
                            }
                        }
                    }
                }
            }

            // 根据物理规律，剔除或修正不合理的数据
            val data = mDataList[i]
            // 1. PM2.5 应该始终小于PM10
            val pm25 = data.getFactorData(FactorType.PM25)
            val pm10 = data.getFactorData(FactorType.PM10)
            if (pm25 != null && pm10 != null) {
                // 若pm2.5大于pm10
                if (pm25 >= pm10){
                    val lastIndex = i - 1
                    // 则将pm2.5修改为前一个数据的值
                    if (lastIndex >= 0) {
                        data.setFactorData(FactorType.PM25, mDataList[lastIndex].getFactorData(FactorType.PM25))
                    } else {
                        if (lastData.isEmpty()) {
                            // 没有历史数据时，修改为pm10的80%(后续待优化比例 2026.3.6)
                            data.setFactorData(FactorType.PM25, data.getFactorData(FactorType.PM10)?.times(.8))
                        } else {
                            data.setFactorData(FactorType.PM25, lastData.last().getFactorData(FactorType.PM25))
                        }
                    }
                }
            }

            i++
        }

        // 将新数据的至多最后15个保存下来（已经过预处理），用于下一次的判断
        val newList = mutableListOf<DataVo>()
        val s = if ((mDataList.lastIndex - ncal + 1) < 0) 0 else mDataList.lastIndex - ncal + 1
        mDataList.subList(s, mDataList.lastIndex + 1).forEach {
            newList.add(it.copy())
        }
        // 当新数据与旧数据采样时间差超过1分钟时，认为两组数据已无关联性，清空旧数据
        if (lastData.isNotEmpty() && newList.isNotEmpty()) {
            val lastTime = DateUtil.instance.StringToDate(lastData.last().time)
            val thisTime = DateUtil.instance.StringToDate(newList.first().time)
            if ((thisTime?.time?.minus(lastTime?.time ?: 0) ?: 0) >= (60 * 1000)) {
                lastData.clear()
            }
        }
        lastData.addAll(newList)
        // 确保保存的数据最多只有最新的15个
        while (lastData.size > ncal) {
            lastData.removeAt(0)
        }

        return mDataList
    }

    /**
@@ -103,15 +255,17 @@
     * @param list 监测数据
     * @return 均值和有效数据个数
     */
    private fun average(list: List<DataVo>, factorName:String?): Pair<Double, Int> {
    private fun average(list: List<DataVo>, factorName: String?): Pair<Double, Int>? {
        var t = 0.0
        var c = 0
        list.forEach {
            for (i in it.values?.indices ?: 0..0) {
                val f = it.values?.get(i)
                if (f?.factorName == factorName) {
                    if (f?.factorData != null) {
                        t += f.factorData!!
                    val range = FactorType.getRange(f?.factorName) ?: continue
                    //判断数据是否在合理范围内
                    if ((f?.factorData ?: 0.0) in range.first..range.second) {
                        t += f?.factorData!!
                        c++
                    }
                    break
@@ -122,10 +276,14 @@
        val avg = if (c == 0) {
            0.0
        } else {
            t / c
            round(t / c * 1000) / 1000
        }

        return Pair(avg, c)
        return if (c == 0) {
            null
        } else {
            Pair(avg, c)
        }
    }

    /**
@@ -138,8 +296,9 @@
            for (i in it.values?.indices ?: 0..0) {
                val f = it.values?.get(i)
                if (f?.factorName == factorName) {
                    if (f?.factorData != null) {
                        t += (f.factorData!! - avg) * (f.factorData!! - avg)
                    val range = FactorType.getRange(f?.factorName) ?: continue
                    if ((f?.factorData ?: 0.0) in range.first..range.second) {
                        t += (f?.factorData!! - avg) * (f.factorData!! - avg)
                        c++
                    }
                    break
@@ -153,4 +312,32 @@
            sqrt(t / (c - 1))
        }
    }

    /**
     * 判断数据是否在正常量程内
     */
    private fun isInRange(airData: AirData): Boolean? {
        val range = FactorType.getRange(airData.factorName) ?: return null
        //判断数据是否在合理范围内
        return (airData.factorData ?: 0.0) in range.first..range.second
    }

    /**
     * 判断连续的数据量级上升幅度是否过大
     */
    private fun excessiveChange(airData: AirData): Boolean? {
        airData.factorData ?: return null
        if (lastData.isEmpty()) return false
        val latestData = lastData.last()
        // 结果倍率
        var m = 1.0
        for (i in latestData.values?.indices ?: 0..0) {
            val f = latestData.values?.get(i)
            if (f?.factorName == airData.factorName) {
                m = airData.factorData!!.div(f?.factorData ?: airData.factorData!!)
                break
            }
        }
        return m > FactorType.getMultiplier(airData.factorName)
    }
}