fix: 时间段查询

snowy-plugin/snowy-plugin-coldchain/src/main/java/vip/xiaonuo/coldchain/core/service/FluxAggregationUtils.java

@@ -0,0 +1,52 @@
+package vip.xiaonuo.coldchain.core.service;
+
+/**
+ * @author jackzhou
+ * @version 1.0
+ * @project jfcloud-coldchain
+ * @description
+ * @date 2024/12/10 12:00:28
+ */
+import lombok.experimental.UtilityClass;
+
+import java.time.Duration;
+import java.time.Instant;
+@UtilityClass
+
+public class FluxAggregationUtils {
+
+    /**
+     * 确定适合的聚合窗口
+     *
+     * @param startTime 起始时间（ISO 8601 格式，例如 "2024-01-01T00:00:00Z"）
+     * @param stopTime  结束时间（ISO 8601 格式，例如 "2024-03-01T00:00:00Z"）
+     * @return 聚合窗口大小 ("1mo", "1w", "1d", "1h")
+     */
+    public static String determineAggregationWindow(String startTime, String stopTime) {
+        Instant start = Instant.parse(startTime);
+        Instant stop = Instant.parse(stopTime);
+
+        // 计算时间间隔
+        Duration duration = Duration.between(start, stop);
+        long days = duration.toDays();
+
+        // 根据间隔选择聚合窗口
+        if (days > 90) {
+            return "1mo"; // 超过 90 天按月度聚合
+        } else if (days > 30) {
+            return "1w"; // 超过 30 天按每周聚合
+        } else if (days > 1) {
+            return "1d"; // 超过 1 天但少于 30 天按每日聚合
+        } else {
+            return "1h"; // 小于或等于 1 天按每小时聚合
+        }
+    }
+
+//    public static void main(String[] args) {
+//        // 示例测试
+//        String startTime = "2024-01-01T00:00:00Z";
+//        String stopTime = "2024-04-01T00:00:00Z";
+//        String aggregationWindow = determineAggregationWindow(startTime, stopTime);
+//        System.out.println("聚合窗口: " + aggregationWindow); // 输出: 1mo
+//    }
+}

snowy-plugin/snowy-plugin-coldchain/src/main/java/vip/xiaonuo/coldchain/core/service/FluxQueryBuilder.java

@@ -0,0 +1,79 @@
+package vip.xiaonuo.coldchain.core.service;
+
+import java.util.Map;
+
+public class FluxQueryBuilder {
+
+    /**
+     * 构造查询温度数据的 Flux 查询字符串
+     *
+     * @param bucket      数据所在的 bucket 名称
+     * @param startTime   查询的开始时间（ISO 8601 格式）
+     * @param endTime     查询的结束时间（ISO 8601 格式）
+     * @param measurement 测量名称
+     * @param field       字段名称（如 temperature）
+     * @param filters     额外的过滤条件（标签键值对）
+     * @param interval    聚合时间间隔（如 1d 表示一天）
+     * @return 构造的 Flux 查询字符串
+     */
+    public static String buildRangeQuery(
+            String bucket,
+            String startTime,
+            String endTime,
+            String measurement,
+            String field,
+            Map<String, String> filters,
+            String interval) {
+
+        StringBuilder flux = new StringBuilder();
+
+        // 指定数据来源 bucket
+        flux.append("from(bucket: \"").append(bucket).append("\")\n");
+
+        // 指定时间范围
+        flux.append("  |> range(start: time(v: \"").append(startTime)
+                .append("\"), stop: time(v: \"").append(endTime).append("\"))\n");
+
+        // 基本过滤条件
+        flux.append("  |> filter(fn: (r) => r[\"_measurement\"] == \"")
+                .append(measurement).append("\" and r[\"_field\"] == \"").append(field).append("\"");
+
+        // 动态标签过滤条件
+        if (filters != null && !filters.isEmpty()) {
+            for (Map.Entry<String, String> filter : filters.entrySet()) {
+                flux.append(" and r[\"").append(filter.getKey()).append("\"] == \"")
+                        .append(filter.getValue()).append("\"");
+            }
+        }
+        flux.append(")\n");
+
+        // 聚合窗口
+        if (interval != null && !interval.isEmpty()) {
+            flux.append("  |> aggregateWindow(every: ").append(interval)
+                    .append(", fn: mean, createEmpty: false)\n");
+        }
+        // 按时间升序排序
+        flux.append("  |> sort(columns: [\"_time\"], desc: false)\n");
+        return flux.toString();
+    }
+
+//    public static void main(String[] args) {
+//        // 示例使用
+//        String bucket = "coldchain";
+//        String startTime = "2024-11-01T00:00:00Z";
+//        String endTime = "2024-11-30T23:59:59Z";
+//        String measurement = "sensor_data";
+//        String field = "temperature";
+//
+//        Map<String, String> filters = Map.of(
+//                "device_id", "30067080",
+//                "roads", "1"
+//        );
+//
+//        String interval = "1d"; // 每天计算平均值
+//        // 构造查询
+//        String fluxQuery = buildTemperatureQuery(bucket, startTime, endTime, measurement, field, filters, interval);
+//        // 输出 Flux 查询
+//        System.out.println(fluxQuery);
+//    }
+}

snowy-plugin/snowy-plugin-coldchain/src/main/java/vip/xiaonuo/coldchain/core/service/JfcloudSensorDataService.java

@@ -10,12 +10,12 @@ import org.springframework.stereotype.Service;
 import vip.xiaonuo.coldchain.core.bean.influxdb.SensorData;
 import vip.xiaonuo.coldchain.core.util.DateFormatter;
 
-import java.time.Duration;
 import java.time.Instant;
 import java.time.LocalDateTime;
 import java.time.ZoneOffset;
 import java.time.format.DateTimeFormatter;
 import java.util.List;
+import java.util.Map;
 import java.util.stream.Collectors;
 
 /**
@@ -63,14 +63,16 @@ public class JfcloudSensorDataService extends JfcloudFluxDataService<SensorData>
      * @param roads        路数
      * @param startTimeStr 查询开始时间
      * @param endTimeStr   查询结束时间
+     * @param field   temperature/humidity/co2
      * @return 查询到的传感器数据列表
      */
 
-    public List<SensorData> queryDataByDeviceIdAndRoads(String deviceId, Integer roads, String startTimeStr, String endTimeStr) {
+    public List<SensorData> queryDataByDeviceIdAndRoads(String deviceId, Integer roads, String startTimeStr, String endTimeStr,String field) {
         Assert.notNull(deviceId, "deviceId cannot be null");
         Assert.notNull(roads, "roads cannot be null");
         Assert.notNull(startTimeStr, "startTime cannot be null");
         Assert.notNull(endTimeStr, "endTime cannot be null");
+        Assert.notNull(field, "field cannot be null");
         //替换起始时间和结束时间2024/09/09 这种格式为2024-09-09
         startTimeStr = DateFormatter.replaceDateFormat(startTimeStr);
         endTimeStr = DateFormatter.replaceDateFormat(endTimeStr);
@@ -83,49 +85,19 @@ public class JfcloudSensorDataService extends JfcloudFluxDataService<SensorData>
         }
         String startTimeFormatted = convertToISO8601(startTimeStr);
         String endTimeFormatted = convertToISO8601(endTimeStr);
-        // Determine the aggregation window (by day or by hour)
-        String aggregationWindow = determineAggregationWindow(startTimeFormatted, endTimeFormatted);
-        String query = buildRangeTimeFluxQuery(getBucketName(), deviceId, roads + "", startTimeFormatted, endTimeFormatted, aggregationWindow);
-        // 执行查询
+        String aggregationWindow = FluxAggregationUtils.determineAggregationWindow(startTimeFormatted, endTimeFormatted);
+        String measurement = "sensor_data";
+        Map<String, String> filters = Map.of(
+                "device_id", deviceId,
+                "roads", roads.toString()
+        );
+        String query = FluxQueryBuilder.buildRangeQuery(getBucketName(), startTimeFormatted, endTimeFormatted, measurement, field, filters, aggregationWindow);
         QueryApi queryApi = jfcloudInfluxDBService.getInfluxDBClient().getQueryApi();
         List<FluxTable> results = queryApi.query(query);
-
-        // 将查询结果转换为 SensorData 实体并返回
         return results.stream().flatMap(table -> table.getRecords().stream()).map(fluxRecord -> mapRecordToEntity(fluxRecord, getEntityClass()))  // 转换为 SensorData 实体
                 .collect(Collectors.toList());  // 返回多条数据
     }
 
-    /**
-     * Builds the Flux query dynamically based on the provided parameters.
-     *
-     * @param bucketName The name of the InfluxDB bucket.
-     * @param deviceId   The device ID to filter by.
-     * @param roads      The roads field to filter by.
-     * @param startTime  The start time of the query (ISO format).
-     * @param stopTime   The end time of the query (ISO format).
-     * @return The Flux query as a string.
-     */
-    public static String buildRangeTimeFluxQuery(String bucketName, String deviceId, String roads, String startTime, String stopTime, String aggregationWindow) {
-        return String.format("temperature = from(bucket: \"%s\")\n" + "  |> range(start: time(v: \"%s\"), stop: time(v: \"%s\"))\n" + "  |> filter(fn: (r) => r[\"_measurement\"] == \"sensor_data\" and r[\"_field\"] == \"temperature\" and r[\"device_id\"] == \"%s\" and r[\"roads\"] == \"%s\")\n" + "  |> aggregateWindow(every: %s, fn: mean, createEmpty: false)\n" + "humidity = from(bucket: \"%s\")\n" + "  |> range(start: time(v: \"%s\"), stop: time(v: \"%s\"))\n" + "  |> filter(fn: (r) => r[\"_measurement\"] == \"sensor_data\" and r[\"_field\"] == \"humidity\" and r[\"device_id\"] == \"%s\" and r[\"roads\"] == \"%s\")\n" + "  |> aggregateWindow(every: %s, fn: mean, createEmpty: false)\n" + "co2 = from(bucket: \"%s\")\n" + "  |> range(start: time(v: \"%s\"), stop: time(v: \"%s\"))\n" + "  |> filter(fn: (r) => r[\"_measurement\"] == \"sensor_data\" and r[\"_field\"] == \"co2\" and r[\"device_id\"] == \"%s\" and r[\"roads\"] == \"%s\")\n" + "  |> aggregateWindow(every: %s, fn: mean, createEmpty: false)\n" + "union(tables: [temperature, humidity, co2])\n" + "  |> pivot(rowKey: [\"_time\"], columnKey: [\"_field\"], valueColumn: \"_value\")\n" + "  |> yield(name: \"latest_data\")", bucketName, startTime, stopTime, deviceId, roads, aggregationWindow, bucketName, startTime, stopTime, deviceId, roads, aggregationWindow, bucketName, startTime, stopTime, deviceId, roads, aggregationWindow);
-    }
-
-    public static String determineAggregationWindow(String startTime, String stopTime) {
-        Instant start = Instant.parse(startTime);
-        Instant stop = Instant.parse(stopTime);
-        // Calculate the duration between start and stop
-        Duration duration = Duration.between(start, stop);
-        // If the duration is more than 30 days, return "1w" (weekly aggregation)
-        if (duration.toDays() > 30) {
-            return "1w"; // Weekly aggregation
-        }
-        // If the duration is more than 1 day but less than or equal to 30 days, return "1d" (daily aggregation)
-        else if (duration.toDays() > 1) {
-            return "1d"; // Daily aggregation
-        } else {
-            // If the duration is less than or equal to 1 day, return "1h" (hourly aggregation)
-            return "1h"; // Hourly aggregation
-        }
-    }
 
     /**
      * Converts a time string in "yyyy-MM-dd HH:mm:ss" format to ISO 8601 format.

snowy-plugin/snowy-plugin-coldchain/src/main/java/vip/xiaonuo/coldchain/modular/app/service/AppDeviceService.java

@@ -149,7 +149,22 @@ public class AppDeviceService {
 
     public List<SensorData> queryDataByDeviceIdAndRoads(AppDeviceQueryParams appDeviceQueryParams) {
         Assert.notNull(appDeviceQueryParams, "appDeviceQueryParams cannot be null");
-        return monitorDeviceService.queryDataByDeviceIdAndRoads(appDeviceQueryParams.getSensorCode(), appDeviceQueryParams.getSensorRoute(), appDeviceQueryParams.getStartTime(), appDeviceQueryParams.getEndTime());
+        MonitorTargetRegion oneByDeviceCodeAndSensorNo = monitorTargetRegionService.findOneByDeviceCodeAndSensorNo(appDeviceQueryParams.getSensorCode(), appDeviceQueryParams.getSensorRoute());
+        if (Objects.nonNull(oneByDeviceCodeAndSensorNo)) {
+            //探头类型
+            String sensorType = oneByDeviceCodeAndSensorNo.getSensorType();
+            if(StrUtil.isNotBlank(sensorType) && sensorType.toLowerCase().contains("w")){
+                List<SensorData> temperatures = monitorDeviceService.queryDataByDeviceIdAndRoads(appDeviceQueryParams.getSensorCode(), appDeviceQueryParams.getSensorRoute(), appDeviceQueryParams.getStartTime(), appDeviceQueryParams.getEndTime(), "temperature");
+                return temperatures;
+            }
+            if(StrUtil.isNotBlank(sensorType) && sensorType.toLowerCase().contains("s")){
+                List<SensorData> humiditys = monitorDeviceService.queryDataByDeviceIdAndRoads(appDeviceQueryParams.getSensorCode(), appDeviceQueryParams.getSensorRoute(), appDeviceQueryParams.getStartTime(), appDeviceQueryParams.getEndTime(), "humidity");
+            }
+            if(StrUtil.isNotBlank(sensorType) && sensorType.toLowerCase().contains("c")){
+                List<SensorData> co2s = monitorDeviceService.queryDataByDeviceIdAndRoads(appDeviceQueryParams.getSensorCode(), appDeviceQueryParams.getSensorRoute(), appDeviceQueryParams.getStartTime(), appDeviceQueryParams.getEndTime(), "co2");
+            }
+        }
+        return Collections.emptyList();
     }
 
 
@@ -176,5 +191,4 @@ public class AppDeviceService {
     }
 
 
-
 }

snowy-plugin/snowy-plugin-coldchain/src/main/java/vip/xiaonuo/coldchain/modular/monitordevice/service/MonitorDeviceService.java

@@ -94,7 +94,8 @@ public interface MonitorDeviceService extends IService<MonitorDevice> {
 
     boolean updateLastLoginTimeByDeviceId(Integer deviceId, Date timestamp);
 
-    List<SensorData> queryDataByDeviceIdAndRoads(String deviceId, Integer sensorRoute, String startTime, String endTime);
+    List<SensorData> queryDataByDeviceIdAndRoads(String deviceId, Integer sensorRoute, String startTime, String endTime,String field);
+
 
     SensorData queryLatestDataByDeviceIdAndRoads(String deviceId, Integer sensorRoute);
 

snowy-plugin/snowy-plugin-coldchain/src/main/java/vip/xiaonuo/coldchain/modular/monitordevice/service/impl/MonitorDeviceServiceImpl.java

@@ -12,7 +12,6 @@
  */
 package vip.xiaonuo.coldchain.modular.monitordevice.service.impl;
 
-import cn.dev33.satoken.stp.StpUtil;
 import cn.hutool.core.bean.BeanUtil;
 import cn.hutool.core.collection.CollStreamUtil;
 import cn.hutool.core.util.ObjectUtil;
@@ -46,7 +45,6 @@ import vip.xiaonuo.coldchain.modular.monitordevice.service.MonitorDeviceService;
 import vip.xiaonuo.coldchain.modular.monitordevicetype.entity.CountEntity;
 import vip.xiaonuo.coldchain.modular.monitordevicetype.entity.MonitorDeviceType;
 import vip.xiaonuo.coldchain.modular.monitordevicetype.service.MonitorDeviceTypeService;
-import vip.xiaonuo.coldchain.modular.monitortarget.entity.MonitorTarget;
 import vip.xiaonuo.coldchain.modular.monitortarget.param.TargetStatus;
 import vip.xiaonuo.coldchain.modular.monitortarget.service.MonitorTargetService;
 import vip.xiaonuo.coldchain.modular.monitortargetregion.service.MonitorTargetRegionService;
@@ -312,8 +310,8 @@ public class MonitorDeviceServiceImpl extends ServiceImpl<MonitorDeviceMapper, M
     }
 
     @Override
-    public List<SensorData> queryDataByDeviceIdAndRoads(String deviceId, Integer roads, String startTime, String endTime) {
-        return sensorDataService.queryDataByDeviceIdAndRoads(deviceId, roads, startTime, endTime);
+    public List<SensorData> queryDataByDeviceIdAndRoads(String deviceId, Integer roads, String startTime, String endTime, String field) {
+        return sensorDataService.queryDataByDeviceIdAndRoads(deviceId, roads, startTime, endTime,field);
     }
 
     @Override