package com.flightfeather.uav.biz.dataanalysis
|
|
import com.flightfeather.uav.biz.FactorFilter
|
import com.flightfeather.uav.domain.entity.BaseRealTimeData
|
import com.flightfeather.uav.socket.eunm.FactorType
|
import java.time.Duration
|
|
/**
|
* 连续类型的异常分析基类,适用于当前数据与相邻数据之间有关联关系的情况
|
*/
|
abstract class BaseExceptionContinuous<V : BaseAnalysisConfig, Y : BaseExceptionResult>(config: V) :
|
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>()
|
|
// 起始数据与末尾数据间隔
|
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)
|
}
|
|
/**
|
* 判断是否满足异常条件
|
*/
|
abstract fun judgeException(p: BaseRealTimeData?, n: BaseRealTimeData): MutableMap<FactorType, Boolean>
|
|
/**
|
* 判断异常出现的连续个数是否满足条件
|
* @param tag 异常数据对象
|
*/
|
abstract fun judgeExceptionCount(tag: Tag): Boolean
|
|
/**
|
* 异常数据的截取判断
|
* 是否需要限制一组异常数据的长度
|
* @return 默认不需要截取
|
*/
|
open fun needCut(tag: Tag): Boolean {
|
return false
|
}
|
|
override fun init() {
|
super.init()
|
lastData = null
|
tagMap.clear()
|
config.factorFilter.mainList().forEach { f ->
|
tagMap[f] = Tag()
|
}
|
}
|
|
override fun onNextData(data: BaseRealTimeData) {
|
val isContinue = isContinuous(lastData, data)
|
val hasException = judgeException(lastData, data)
|
config.factorFilter.selectedList.forEach { s ->
|
val f = s.main
|
tagMap[f]?.let {
|
it.eIndex++
|
// 起始数据
|
it.endData = data
|
if (it.startData == null) {
|
it.refreshAfterCheckResult(data)
|
}
|
// 判断相邻数据是否连续并且是否满足异常判断
|
if (!isContinue || needCut(it)) {
|
checkResult(s)
|
// 数据不连续时，记录异常情况
|
if (it.eIndex - it.sIndex >= durationCount) {
|
it.refreshAfterCheckResult(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)
|
}
|
}
|
}
|
}
|
}
|
lastData = data
|
}
|
|
override fun onDone() {
|
checkResult()
|
}
|
|
/**
|
* 检查连续异常结束时，是否符合异常存储条件
|
*/
|
open fun checkResult(factor: FactorFilter.SelectedFactor? = null) {
|
val tag = tagMap[factor?.main]
|
if (factor != null && tag != null) {
|
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
|
}
|
|
/**
|
* 生成一条异常分析结果
|
*/
|
abstract fun newResult(
|
start: BaseRealTimeData,
|
end: BaseRealTimeData?,
|
factor: FactorFilter.SelectedFactor,
|
exceptionData: List<BaseRealTimeData>,
|
): Y
|
|
}
|
