package com.dingzhuo.energy.dataservice.data;
|
|
import com.dingzhuo.energy.dataservice.data.influxdb.FastDB;
|
import com.dingzhuo.energy.dataservice.data.influxdb.QueryType;
|
import com.dingzhuo.energy.dataservice.data.influxdb.RtdbResult;
|
import com.dingzhuo.energy.dataservice.domain.CollectionModes;
|
import com.dingzhuo.energy.dataservice.domain.Quality;
|
import com.dingzhuo.energy.dataservice.domain.RetrievalModes;
|
import com.dingzhuo.energy.dataservice.domain.TagValue;
|
import com.dingzhuo.energy.dataservice.service.RealtimeDatabase;
|
import com.google.common.base.Function;
|
import org.apache.logging.log4j.LogManager;
|
import org.apache.logging.log4j.Logger;
|
import org.joda.time.DateTime;
|
|
import java.time.ZoneOffset;
|
import java.util.ArrayList;
|
import java.util.Date;
|
import java.util.List;
|
import java.util.Objects;
|
import java.util.stream.Collectors;
|
|
/**
|
* @author fanxinfu
|
*/
|
public class InfluxDb implements RealtimeDatabase {
|
private Logger logger = LogManager.getLogger(InfluxDb.class);
|
|
private FastDB fastDB;
|
private static final String DB_NAME = "daq";
|
|
/**
|
* 关闭连接
|
*/
|
@Override
|
public void close() {
|
if (!Objects.isNull(fastDB)) {
|
fastDB.close();
|
}
|
}
|
|
/**
|
* 根据点位号获取实时数据
|
*
|
* @param tagCodes 点位号列表
|
* @return
|
*/
|
@Override
|
public List<TagValue> retrieve(List<String> tagCodes) throws Exception {
|
List<RtdbResult> results = fastDB.snapShot(tagCodes);
|
return convertTagValue(results);
|
}
|
|
/**
|
* 根据点位号获取某一时刻的历史数据
|
*
|
* @param tagCodes 点位号集合
|
* @param dataTime 历史时刻
|
* @param timeCode 区分时间类型的time code
|
* @return
|
*/
|
@Override
|
public List<TagValue> retrieve(List<String> tagCodes, Date dataTime, String timeCode) throws Exception {
|
List<RtdbResult> results = fastDB.getDataByTime(tagCodes, dataTime, timeCode);
|
return convertTagValue(results);
|
}
|
|
/**
|
* 根据查询方式获取一段时间内的历史数据
|
*
|
* @param tagCodes 点位号集合
|
* @param beginTime 开始时间
|
* @param endTime 结束时间
|
* @param retrievalModes 查询方式（循环、拟合、全部）
|
* @param pointCount 要查询的数据个数
|
* @return
|
*/
|
@Override
|
public List<TagValue> retrieve(List<String> tagCodes, Date beginTime, Date endTime, RetrievalModes retrievalModes, int pointCount) throws Exception {
|
QueryType queryType = null;
|
switch (retrievalModes) {
|
case BestFit:
|
queryType = QueryType.FIRST;
|
break;
|
default:
|
queryType = QueryType.FULL;
|
break;
|
}
|
List<RtdbResult> results = fastDB.getHistoryData(tagCodes, beginTime, endTime, queryType, pointCount);
|
return convertTagValue(results);
|
}
|
|
/**
|
* 汇总实时数据
|
*
|
* @param tagCodes 点位号集合
|
* @param beginTime 开始时间
|
* @param endTime 结束时间
|
* @param collectionModes 汇总方式（最大值、最小值、平均值、求和等）
|
* @return
|
*/
|
@Override
|
public List<TagValue> statistics(List<String> tagCodes, Date beginTime, Date endTime, CollectionModes collectionModes) throws Exception {
|
DateTime begin = new DateTime(beginTime).withMillisOfSecond(0);
|
DateTime end = new DateTime(endTime).withMillisOfSecond(0);
|
QueryType queryType = null;
|
switch (collectionModes) {
|
case Sum:
|
queryType = QueryType.SUM;
|
break;
|
case Maximum:
|
queryType = QueryType.MAX;
|
break;
|
case Minimum:
|
queryType = QueryType.MIN;
|
break;
|
case Mean:
|
queryType = QueryType.MEAN;
|
break;
|
case Integral:
|
queryType = QueryType.INTEGRAL;
|
break;
|
default:
|
return new ArrayList<>();
|
}
|
List<RtdbResult> results = fastDB.statistics(tagCodes, begin.toDate(), end.toDate(), queryType);
|
return convertTagValue(results);
|
}
|
|
/**
|
* 插入实时数据
|
*
|
* @param tagValues 实时数据集合
|
* @return
|
*/
|
@Override
|
public Boolean storeData(List<TagValue> tagValues) throws Exception {
|
List<RtdbResult> results = convertToRtdbResult(tagValues);
|
return fastDB.storeData(results);
|
}
|
|
/**
|
* 插入历史数据
|
*
|
* @param tagValues 历史数据集合
|
* @return
|
*/
|
@Override
|
public Boolean insertData(List<TagValue> tagValues) throws Exception {
|
List<RtdbResult> results = convertToRtdbResult(tagValues);
|
return fastDB.insertData(results);
|
}
|
|
/**
|
* 打开连接
|
*
|
* @param host 实时数据库地址
|
* @param port 端口号
|
* @param userName 登录用户名
|
* @param pwd 登录密码
|
* @return 是否连接成功
|
*/
|
@Override
|
public boolean open(String host, int port, String userName, String pwd) {
|
try {
|
fastDB = new FastDB(host, port, DB_NAME);
|
} catch (Throwable ex) {
|
ex.printStackTrace();
|
}
|
return true;
|
}
|
|
|
private List<TagValue> convertTagValue(List<RtdbResult> results) {
|
Function<RtdbResult, TagValue> convertFun = input -> {
|
TagValue tagValue = new TagValue();
|
tagValue.setDataTime(Date.from(input.getTime().atZone(ZoneOffset.ofHours(8)).toInstant()));
|
tagValue.setQuality(input.getQuality() > 0 ? Quality.GOOD : Quality.BAD);
|
tagValue.setTagCode(input.getTagCode());
|
tagValue.setValue(input.getValue());
|
if (tagValue.getQuality().equals(Quality.BAD)) {
|
logger.info(input.toString());
|
logger.info(tagValue.toString());
|
}
|
return tagValue;
|
};
|
return results.stream().map(convertFun).collect(Collectors.toList());
|
}
|
|
private List<RtdbResult> convertToRtdbResult(List<TagValue> results) {
|
Function<TagValue, RtdbResult> convertFun = input -> {
|
RtdbResult rtdbResult = new RtdbResult();
|
rtdbResult.setTime(input.getDataTime().toInstant());
|
rtdbResult.setQuality(input.getQuality() == Quality.GOOD ? 1 : 0);
|
rtdbResult.setTagCode(input.getTagCode());
|
rtdbResult.setValue(input.getValue());
|
return rtdbResult;
|
};
|
return results.stream().map(convertFun).collect(Collectors.toList());
|
}
|
}
|
