|
|
@@ -1,170 +1,196 @@
|
|
|
-package com.usky.vpp.service.impl;
|
|
|
-
|
|
|
-import com.usky.common.core.bean.ApiResult;
|
|
|
-import com.usky.demo.RemoteTsdbProxyService;
|
|
|
-import com.usky.demo.domain.HistorysInnerRequestVO;
|
|
|
-import com.usky.demo.domain.HistorysInnerResultVO;
|
|
|
-import com.usky.demo.domain.LastInnerQueryVO;
|
|
|
-import com.usky.demo.domain.LastInnerResultVO;
|
|
|
-import com.usky.demo.domain.MetricVO;
|
|
|
-import com.usky.vpp.constant.VppTsdbConstants;
|
|
|
-import com.usky.vpp.service.VppTsdbQueryService;
|
|
|
-import org.slf4j.Logger;
|
|
|
-import org.slf4j.LoggerFactory;
|
|
|
-import org.springframework.beans.factory.annotation.Autowired;
|
|
|
-import org.springframework.stereotype.Service;
|
|
|
-import org.springframework.util.CollectionUtils;
|
|
|
-import org.springframework.util.StringUtils;
|
|
|
-
|
|
|
-import java.math.BigDecimal;
|
|
|
-import java.math.RoundingMode;
|
|
|
-import java.time.LocalDateTime;
|
|
|
-import java.time.format.DateTimeFormatter;
|
|
|
-import java.time.format.DateTimeFormatterBuilder;
|
|
|
-import java.time.temporal.ChronoField;
|
|
|
-import java.util.ArrayList;
|
|
|
-import java.util.Collections;
|
|
|
-import java.util.HashMap;
|
|
|
-import java.util.LinkedHashSet;
|
|
|
-import java.util.List;
|
|
|
-import java.util.Map;
|
|
|
-import java.util.TreeMap;
|
|
|
-
|
|
|
-/**
|
|
|
- * TDengine 时序数据查询(通过 service-tsdb Feign,与 service-ems 一致)
|
|
|
- */
|
|
|
-@Service
|
|
|
-public class VppTsdbQueryServiceImpl implements VppTsdbQueryService {
|
|
|
-
|
|
|
- private static final Logger log = LoggerFactory.getLogger(VppTsdbQueryServiceImpl.class);
|
|
|
-
|
|
|
- private static final DateTimeFormatter TSDB_FLEXIBLE_TIME_FORMAT = new DateTimeFormatterBuilder()
|
|
|
- .appendPattern("yyyy-MM-dd HH:mm:ss")
|
|
|
- .optionalStart()
|
|
|
- .appendFraction(ChronoField.NANO_OF_SECOND, 0, 9, true)
|
|
|
- .optionalEnd()
|
|
|
- .toFormatter();
|
|
|
-
|
|
|
- @Autowired
|
|
|
- private RemoteTsdbProxyService remoteTsdbProxyService;
|
|
|
-
|
|
|
- @Override
|
|
|
- public Map<String, Map<String, TreeMap<LocalDateTime, BigDecimal>>> queryDeviceMetricHistory(
|
|
|
- List<String> deviceUuids,
|
|
|
- LocalDateTime startTime,
|
|
|
- LocalDateTime endTime,
|
|
|
- List<String> metrics) {
|
|
|
- if (CollectionUtils.isEmpty(deviceUuids) || startTime == null || endTime == null) {
|
|
|
- return Collections.emptyMap();
|
|
|
- }
|
|
|
-
|
|
|
- HistorysInnerRequestVO request = new HistorysInnerRequestVO();
|
|
|
- request.setDeviceuuid(new ArrayList<>(new LinkedHashSet<>(deviceUuids)));
|
|
|
- request.setStartTime(startTime.format(DateTimeFormatter.ofPattern(VppTsdbConstants.TIME_FORMAT)));
|
|
|
- request.setEndTime(endTime.format(DateTimeFormatter.ofPattern(VppTsdbConstants.TIME_FORMAT)));
|
|
|
- if (!CollectionUtils.isEmpty(metrics)) {
|
|
|
- request.setMetrics(new ArrayList<>(metrics));
|
|
|
- }
|
|
|
-
|
|
|
- try {
|
|
|
- ApiResult<List<HistorysInnerResultVO>> result = remoteTsdbProxyService.queryHistoryDeviceData(request);
|
|
|
- if (result == null || !result.isSuccess() || CollectionUtils.isEmpty(result.getData())) {
|
|
|
- log.warn("TDengine 历史查询无数据, devices={}, start={}, end={}",
|
|
|
- deviceUuids.size(), request.getStartTime(), request.getEndTime());
|
|
|
- return Collections.emptyMap();
|
|
|
- }
|
|
|
- return parseHistoryResult(result.getData());
|
|
|
- } catch (Exception ex) {
|
|
|
- log.error("TDengine 历史查询失败: {}", ex.getMessage(), ex);
|
|
|
- return Collections.emptyMap();
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- @Override
|
|
|
- public Map<String, Map<String, BigDecimal>> queryLatestMetrics(List<String> deviceUuids, List<String> metrics) {
|
|
|
- if (CollectionUtils.isEmpty(deviceUuids)) {
|
|
|
- return Collections.emptyMap();
|
|
|
- }
|
|
|
-
|
|
|
- LastInnerQueryVO request = new LastInnerQueryVO();
|
|
|
- request.setDeviceuuid(new ArrayList<>(new LinkedHashSet<>(deviceUuids)));
|
|
|
- if (!CollectionUtils.isEmpty(metrics)) {
|
|
|
- request.setMetrics(new ArrayList<>(metrics));
|
|
|
- }
|
|
|
-
|
|
|
- try {
|
|
|
- ApiResult<List<LastInnerResultVO>> result = remoteTsdbProxyService.queryLastDeviceData(request);
|
|
|
- if (result == null || !result.isSuccess() || CollectionUtils.isEmpty(result.getData())) {
|
|
|
- return Collections.emptyMap();
|
|
|
- }
|
|
|
- return parseLatestResult(result.getData());
|
|
|
- } catch (Exception ex) {
|
|
|
- log.error("TDengine 最新值查询失败: {}", ex.getMessage(), ex);
|
|
|
- return Collections.emptyMap();
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- private Map<String, Map<String, TreeMap<LocalDateTime, BigDecimal>>> parseHistoryResult(
|
|
|
- List<HistorysInnerResultVO> historyDataList) {
|
|
|
- Map<String, Map<String, TreeMap<LocalDateTime, BigDecimal>>> deviceMetricMap = new HashMap<>();
|
|
|
- for (HistorysInnerResultVO result : historyDataList) {
|
|
|
- if (result == null || !StringUtils.hasText(result.getDeviceuuid()) || CollectionUtils.isEmpty(result.getMetrics())) {
|
|
|
- continue;
|
|
|
- }
|
|
|
- String deviceUuid = result.getDeviceuuid();
|
|
|
- Map<String, TreeMap<LocalDateTime, BigDecimal>> metricMap =
|
|
|
- deviceMetricMap.computeIfAbsent(deviceUuid, key -> new HashMap<>());
|
|
|
-
|
|
|
- for (MetricVO metric : result.getMetrics()) {
|
|
|
- if (metric == null || !StringUtils.hasText(metric.getMetric()) || CollectionUtils.isEmpty(metric.getMetricItems())) {
|
|
|
- continue;
|
|
|
- }
|
|
|
- TreeMap<LocalDateTime, BigDecimal> timeValueMap =
|
|
|
- metricMap.computeIfAbsent(metric.getMetric(), key -> new TreeMap<>());
|
|
|
- for (Map<String, Object> point : metric.getMetricItems()) {
|
|
|
- if (point == null || point.get("timestamp") == null || point.get("value") == null) {
|
|
|
- continue;
|
|
|
- }
|
|
|
- try {
|
|
|
- LocalDateTime timestamp = parseTimestamp(point.get("timestamp").toString());
|
|
|
- BigDecimal value = new BigDecimal(point.get("value").toString());
|
|
|
- timeValueMap.put(timestamp, value);
|
|
|
- } catch (Exception ignored) {
|
|
|
- // skip invalid point
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- return deviceMetricMap;
|
|
|
- }
|
|
|
-
|
|
|
- private Map<String, Map<String, BigDecimal>> parseLatestResult(List<LastInnerResultVO> latestList) {
|
|
|
- Map<String, Map<String, BigDecimal>> result = new HashMap<>();
|
|
|
- for (LastInnerResultVO item : latestList) {
|
|
|
- if (item == null || !StringUtils.hasText(item.getDeviceuuid()) || item.getMetrics() == null) {
|
|
|
- continue;
|
|
|
- }
|
|
|
- Map<String, BigDecimal> metricValues = new HashMap<>();
|
|
|
- item.getMetrics().forEach((metric, value) -> {
|
|
|
- if (value == null) {
|
|
|
- return;
|
|
|
- }
|
|
|
- try {
|
|
|
- metricValues.put(metric, new BigDecimal(value.toString()).setScale(4, RoundingMode.HALF_UP));
|
|
|
- } catch (Exception ignored) {
|
|
|
- // skip invalid value
|
|
|
- }
|
|
|
- });
|
|
|
- result.put(item.getDeviceuuid(), metricValues);
|
|
|
- }
|
|
|
- return result;
|
|
|
- }
|
|
|
-
|
|
|
- private LocalDateTime parseTimestamp(String timestampStr) {
|
|
|
- String normalized = timestampStr.trim();
|
|
|
- if (normalized.endsWith(".0")) {
|
|
|
- normalized = normalized.substring(0, normalized.length() - 2);
|
|
|
- }
|
|
|
- return LocalDateTime.parse(normalized, TSDB_FLEXIBLE_TIME_FORMAT);
|
|
|
- }
|
|
|
-}
|
|
|
+package com.usky.vpp.service.impl;
|
|
|
+
|
|
|
+import com.usky.common.core.bean.ApiResult;
|
|
|
+import com.usky.demo.RemoteTsdbProxyService;
|
|
|
+import com.usky.demo.domain.HistorysInnerRequestVO;
|
|
|
+import com.usky.demo.domain.HistorysInnerResultVO;
|
|
|
+import com.usky.demo.domain.LastInnerQueryVO;
|
|
|
+import com.usky.demo.domain.LastInnerResultVO;
|
|
|
+import com.usky.demo.domain.MetricVO;
|
|
|
+import com.usky.vpp.constant.VppTsdbConstants;
|
|
|
+import com.usky.vpp.service.VppTsdbQueryService;
|
|
|
+import com.usky.vpp.util.VppBaselineHelper;
|
|
|
+import org.slf4j.Logger;
|
|
|
+import org.slf4j.LoggerFactory;
|
|
|
+import org.springframework.beans.factory.annotation.Autowired;
|
|
|
+import org.springframework.stereotype.Service;
|
|
|
+import org.springframework.util.CollectionUtils;
|
|
|
+import org.springframework.util.StringUtils;
|
|
|
+
|
|
|
+import java.math.BigDecimal;
|
|
|
+import java.math.RoundingMode;
|
|
|
+import java.time.LocalDate;
|
|
|
+import java.time.LocalDateTime;
|
|
|
+import java.time.LocalTime;
|
|
|
+import java.time.format.DateTimeFormatter;
|
|
|
+import java.time.format.DateTimeFormatterBuilder;
|
|
|
+import java.time.temporal.ChronoField;
|
|
|
+import java.util.ArrayList;
|
|
|
+import java.util.Collections;
|
|
|
+import java.util.HashMap;
|
|
|
+import java.util.LinkedHashSet;
|
|
|
+import java.util.List;
|
|
|
+import java.util.Map;
|
|
|
+import java.util.TreeMap;
|
|
|
+
|
|
|
+/**
|
|
|
+ * TDengine 时序数据查询(通过 service-tsdb Feign,与 service-ems 一致)
|
|
|
+ */
|
|
|
+@Service
|
|
|
+public class VppTsdbQueryServiceImpl implements VppTsdbQueryService {
|
|
|
+
|
|
|
+ private static final Logger log = LoggerFactory.getLogger(VppTsdbQueryServiceImpl.class);
|
|
|
+
|
|
|
+ private static final DateTimeFormatter TSDB_FLEXIBLE_TIME_FORMAT = new DateTimeFormatterBuilder()
|
|
|
+ .appendPattern("yyyy-MM-dd HH:mm:ss")
|
|
|
+ .optionalStart()
|
|
|
+ .appendFraction(ChronoField.NANO_OF_SECOND, 0, 9, true)
|
|
|
+ .optionalEnd()
|
|
|
+ .toFormatter();
|
|
|
+
|
|
|
+ @Autowired
|
|
|
+ private RemoteTsdbProxyService remoteTsdbProxyService;
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public Map<String, Map<String, TreeMap<LocalDateTime, BigDecimal>>> queryDeviceMetricHistory(
|
|
|
+ List<String> deviceUuids,
|
|
|
+ LocalDateTime startTime,
|
|
|
+ LocalDateTime endTime,
|
|
|
+ List<String> metrics) {
|
|
|
+ if (CollectionUtils.isEmpty(deviceUuids) || startTime == null || endTime == null) {
|
|
|
+ return Collections.emptyMap();
|
|
|
+ }
|
|
|
+
|
|
|
+ HistorysInnerRequestVO request = new HistorysInnerRequestVO();
|
|
|
+ request.setDeviceuuid(new ArrayList<>(new LinkedHashSet<>(deviceUuids)));
|
|
|
+ request.setStartTime(startTime.format(DateTimeFormatter.ofPattern(VppTsdbConstants.TIME_FORMAT)));
|
|
|
+ request.setEndTime(endTime.format(DateTimeFormatter.ofPattern(VppTsdbConstants.TIME_FORMAT)));
|
|
|
+ if (!CollectionUtils.isEmpty(metrics)) {
|
|
|
+ request.setMetrics(new ArrayList<>(metrics));
|
|
|
+ }
|
|
|
+
|
|
|
+ try {
|
|
|
+ ApiResult<List<HistorysInnerResultVO>> result = remoteTsdbProxyService.queryHistoryDeviceData(request);
|
|
|
+ if (result == null || !result.isSuccess() || CollectionUtils.isEmpty(result.getData())) {
|
|
|
+ log.warn("TDengine 历史查询无数据, devices={}, start={}, end={}",
|
|
|
+ deviceUuids.size(), request.getStartTime(), request.getEndTime());
|
|
|
+ return Collections.emptyMap();
|
|
|
+ }
|
|
|
+ return parseHistoryResult(result.getData());
|
|
|
+ } catch (Exception ex) {
|
|
|
+ log.error("TDengine 历史查询失败: {}", ex.getMessage(), ex);
|
|
|
+ return Collections.emptyMap();
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public Map<String, Map<String, BigDecimal>> queryLatestMetrics(List<String> deviceUuids, List<String> metrics) {
|
|
|
+ if (CollectionUtils.isEmpty(deviceUuids)) {
|
|
|
+ return Collections.emptyMap();
|
|
|
+ }
|
|
|
+
|
|
|
+ LastInnerQueryVO request = new LastInnerQueryVO();
|
|
|
+ request.setDeviceuuid(new ArrayList<>(new LinkedHashSet<>(deviceUuids)));
|
|
|
+ if (!CollectionUtils.isEmpty(metrics)) {
|
|
|
+ request.setMetrics(new ArrayList<>(metrics));
|
|
|
+ }
|
|
|
+
|
|
|
+ try {
|
|
|
+ ApiResult<List<LastInnerResultVO>> result = remoteTsdbProxyService.queryLastDeviceData(request);
|
|
|
+ if (result == null || !result.isSuccess() || CollectionUtils.isEmpty(result.getData())) {
|
|
|
+ return Collections.emptyMap();
|
|
|
+ }
|
|
|
+ return parseLatestResult(result.getData());
|
|
|
+ } catch (Exception ex) {
|
|
|
+ log.error("TDengine 最新值查询失败: {}", ex.getMessage(), ex);
|
|
|
+ return Collections.emptyMap();
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public Map<String, BigDecimal> queryBaselineAvgByTimePoint(List<String> deviceUuids,
|
|
|
+ List<LocalDate> referenceDates,
|
|
|
+ LocalTime periodStart,
|
|
|
+ LocalTime periodEnd,
|
|
|
+ List<String> metrics,
|
|
|
+ Integer intervalMinutes,
|
|
|
+ BigDecimal correctionFactor) {
|
|
|
+ if (CollectionUtils.isEmpty(deviceUuids) || CollectionUtils.isEmpty(referenceDates)) {
|
|
|
+ return Collections.emptyMap();
|
|
|
+ }
|
|
|
+ int interval = intervalMinutes != null && intervalMinutes > 0
|
|
|
+ ? intervalMinutes : VppBaselineHelper.INTERVAL_MINUTES;
|
|
|
+ return VppBaselineHelper.aggregateBaselineAvgByTimePoint(
|
|
|
+ referenceDates,
|
|
|
+ periodStart,
|
|
|
+ periodEnd,
|
|
|
+ interval,
|
|
|
+ correctionFactor,
|
|
|
+ (start, end) -> queryDeviceMetricHistory(deviceUuids, start, end, metrics));
|
|
|
+ }
|
|
|
+
|
|
|
+ private Map<String, Map<String, TreeMap<LocalDateTime, BigDecimal>>> parseHistoryResult(
|
|
|
+ List<HistorysInnerResultVO> historyDataList) {
|
|
|
+ Map<String, Map<String, TreeMap<LocalDateTime, BigDecimal>>> deviceMetricMap = new HashMap<>();
|
|
|
+ for (HistorysInnerResultVO result : historyDataList) {
|
|
|
+ if (result == null || !StringUtils.hasText(result.getDeviceuuid()) || CollectionUtils.isEmpty(result.getMetrics())) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ String deviceUuid = result.getDeviceuuid();
|
|
|
+ Map<String, TreeMap<LocalDateTime, BigDecimal>> metricMap =
|
|
|
+ deviceMetricMap.computeIfAbsent(deviceUuid, key -> new HashMap<>());
|
|
|
+
|
|
|
+ for (MetricVO metric : result.getMetrics()) {
|
|
|
+ if (metric == null || !StringUtils.hasText(metric.getMetric()) || CollectionUtils.isEmpty(metric.getMetricItems())) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ TreeMap<LocalDateTime, BigDecimal> timeValueMap =
|
|
|
+ metricMap.computeIfAbsent(metric.getMetric(), key -> new TreeMap<>());
|
|
|
+ for (Map<String, Object> point : metric.getMetricItems()) {
|
|
|
+ if (point == null || point.get("timestamp") == null || point.get("value") == null) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ try {
|
|
|
+ LocalDateTime timestamp = parseTimestamp(point.get("timestamp").toString());
|
|
|
+ BigDecimal value = new BigDecimal(point.get("value").toString());
|
|
|
+ timeValueMap.put(timestamp, value);
|
|
|
+ } catch (Exception ignored) {
|
|
|
+ // skip invalid point
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return deviceMetricMap;
|
|
|
+ }
|
|
|
+
|
|
|
+ private Map<String, Map<String, BigDecimal>> parseLatestResult(List<LastInnerResultVO> latestList) {
|
|
|
+ Map<String, Map<String, BigDecimal>> result = new HashMap<>();
|
|
|
+ for (LastInnerResultVO item : latestList) {
|
|
|
+ if (item == null || !StringUtils.hasText(item.getDeviceuuid()) || item.getMetrics() == null) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ Map<String, BigDecimal> metricValues = new HashMap<>();
|
|
|
+ item.getMetrics().forEach((metric, value) -> {
|
|
|
+ if (value == null) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ try {
|
|
|
+ metricValues.put(metric, new BigDecimal(value.toString()).setScale(4, RoundingMode.HALF_UP));
|
|
|
+ } catch (Exception ignored) {
|
|
|
+ // skip invalid value
|
|
|
+ }
|
|
|
+ });
|
|
|
+ result.put(item.getDeviceuuid(), metricValues);
|
|
|
+ }
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ private LocalDateTime parseTimestamp(String timestampStr) {
|
|
|
+ String normalized = timestampStr.trim();
|
|
|
+ if (normalized.endsWith(".0")) {
|
|
|
+ normalized = normalized.substring(0, normalized.length() - 2);
|
|
|
+ }
|
|
|
+ return LocalDateTime.parse(normalized, TSDB_FLEXIBLE_TIME_FORMAT);
|
|
|
+ }
|
|
|
+}
|
|
|
+
|