完成报表数据策略数据按照不同模式读取的定时任务，完成报表数据策略读取数据的存储，提供相应的插入，查询接口

industry-system/cqcy-ei-common/src/main/java/com/example/opc_common/util/ConstantStr.java

@@ -169,10 +169,10 @@ public class ConstantStr {
     public static final String LOW_ALARM = "低报";
     public static final String HIGH_ALARM = "高报";
     public static final String HIGH_HIGH_ALARM = "高高报";
-    public static final String ALARM_REDIS_KEY="alarm_redis_key";
-    public static final String ALARM_REDIS_DISTINCT="alarm_redis_distinct";
+    public static final String ALARM_REDIS_KEY = "alarm_redis_key";
+    public static final String ALARM_REDIS_DISTINCT = "alarm_redis_distinct";
 
     //报警监听频率字典值
-    public static final String ALARM_LISTEN_FRE="alarm_listen_fre";
+    public static final String ALARM_LISTEN_FRE = "alarm_listen_fre";
 
 }

industry-system/cqcy-ei-influxdb/pom.xml

@@ -24,6 +24,11 @@
 
     <dependencies>
         <dependency>
+            <groupId>com.example</groupId>
+            <artifactId>cqcy-ei-common</artifactId>
+            <version>0.0.1-SNAPSHOT</version>
+        </dependency>
+        <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter</artifactId>
         </dependency>

industry-system/cqcy-ei-influxdb/src/main/java/com/cqcy/ei/influxdb/service/InFluxDBService.java

@@ -1,13 +1,18 @@
 package com.cqcy.ei.influxdb.service;
 
+import com.cqcy.ei.influxdb.entity.Item;
 import com.cqcy.ei.influxdb.entity.RangeEnum;
+import com.example.opc_common.entity.ItemParent;
 
+import javax.validation.constraints.NotBlank;
 import java.util.Date;
+import java.util.List;
 
 public interface InFluxDBService {
 
     /**
      * 检查bucket是否存在
+     *
      * @param bucket 名称
      * @return boolean
      */
@@ -15,6 +20,7 @@ public interface InFluxDBService {
 
     /**
      * 根据名称创建库
+     *
      * @param bucket 名称
      * @return boolean
      */
@@ -22,6 +28,7 @@ public interface InFluxDBService {
 
     /**
      * 检查并创建库
+     *
      * @param bucket 名称
      * @return boolean
      */
@@ -29,25 +36,28 @@ public interface InFluxDBService {
 
     /**
      * 删除表数据
-     * @param startTime 开始时间
-     * @param endTime 结束时间
+     *
+     * @param startTime   开始时间
+     * @param endTime     结束时间
      * @param measurement 表
-     * @param bucket 库
+     * @param bucket      库
      * @return boolean
      */
     boolean deleteMeasurementByTime(Date startTime, Date endTime, String measurement, String bucket);
 
     /**
      * 删除表数据
+     *
      * @param startTime 开始时间
-     * @param endTime 结束时间
+     * @param endTime   结束时间
      * @return boolean
      */
     boolean deleteMeasurementByTime(Date startTime, Date endTime);
 
     /**
      * 删除表数据
-     * @param time 操作时间
+     *
+     * @param time      操作时间
      * @param rangeEnum 时间类型
      * @return boolean
      */
@@ -55,21 +65,23 @@ public interface InFluxDBService {
 
     /**
      * 删除表数据
-     * @param time 操作时间
-     * @param rangeEnum 时间类型
+     *
+     * @param time        操作时间
+     * @param rangeEnum   时间类型
      * @param measurement 表
-     * @param bucket 库
+     * @param bucket      库
      * @return boolean
      */
     boolean deleteMeasurementByTime(Integer time, RangeEnum rangeEnum, String measurement, String bucket);
 
     /**
      * 迁移表数据
-     * @param startTime 开始时间
-     * @param endTime 结束时间
-     * @param bucket 源库
-     * @param measurement 源表
-     * @param targetBucket 目标库
+     *
+     * @param startTime         开始时间
+     * @param endTime           结束时间
+     * @param bucket            源库
+     * @param measurement       源表
+     * @param targetBucket      目标库
      * @param targetMeasurement 目标表
      * @return boolean
      */
@@ -77,9 +89,10 @@ public interface InFluxDBService {
 
     /**
      * 迁移表数据
-     * @param startTime 开始时间
-     * @param endTime 结束时间
-     * @param targetBucket 目标库
+     *
+     * @param startTime         开始时间
+     * @param endTime           结束时间
+     * @param targetBucket      目标库
      * @param targetMeasurement 目标表
      * @return boolean
      */
@@ -87,9 +100,10 @@ public interface InFluxDBService {
 
     /**
      * 迁移表数据
-     * @param time 操作时间
-     * @param rangeEnum 时间类型
-     * @param targetBucket 目标库
+     *
+     * @param time              操作时间
+     * @param rangeEnum         时间类型
+     * @param targetBucket      目标库
      * @param targetMeasurement 目标表
      * @return boolean
      */
@@ -97,11 +111,12 @@ public interface InFluxDBService {
 
     /**
      * 迁移表数据
-     * @param time 操作时间
-     * @param rangeEnum 时间类型
-     * @param bucket 源库
-     * @param measurement 源表
-     * @param targetBucket 目标库
+     *
+     * @param time              操作时间
+     * @param rangeEnum         时间类型
+     * @param bucket            源库
+     * @param measurement       源表
+     * @param targetBucket      目标库
      * @param targetMeasurement 目标表
      * @return boolean
      */
@@ -109,10 +124,11 @@ public interface InFluxDBService {
 
     /**
      * 迁移指定时间往前一段时间的表数据
-     * @param startTime 开始时间
-     * @param time 操作时间
-     * @param rangeEnum 时间类型
-     * @param targetBucket 目标库
+     *
+     * @param startTime         开始时间
+     * @param time              操作时间
+     * @param rangeEnum         时间类型
+     * @param targetBucket      目标库
      * @param targetMeasurement 目标表
      * @return boolean
      */
@@ -120,12 +136,13 @@ public interface InFluxDBService {
 
     /**
      * 迁移指定时间往前一段时间的表数据
-     * @param startTime 开始时间
-     * @param time 操作时间
-     * @param rangeEnum 时间类型
-     * @param bucket 源库
-     * @param measurement 源表
-     * @param targetBucket 目标库
+     *
+     * @param startTime         开始时间
+     * @param time              操作时间
+     * @param rangeEnum         时间类型
+     * @param bucket            源库
+     * @param measurement       源表
+     * @param targetBucket      目标库
      * @param targetMeasurement 目标表
      * @return boolean
      */
@@ -133,9 +150,38 @@ public interface InFluxDBService {
 
     /**
      * 查询源表是否含有历史数据
+     *
      * @param startTime 开始时间
-     * @param endTime 结束时间
+     * @param endTime   结束时间
      * @return Integer
      */
     Integer getMeasurementCountByTime(Date startTime, Date endTime);
+
+    /**
+     * 检查库并创建表
+     * @param bucket 名称
+     * @return boolean
+     */
+    /**
+     *
+     * @param bucket
+     * @param measurement
+     * @return
+     */
+//    boolean checkAndCreateMeasurement(String bucket,String measurement);
+
+    /**
+     * 批量插入数据
+     *
+     * @param bucket         数据库名
+     * @param measurement    表名
+     * @param itemParentList 过滤数据项集合信息
+     * @return
+     */
+    Boolean batchInsert(String bucket, String measurement,  List<? extends ItemParent> itemParentList);
+
+
+    List<Item> queryLast(String bucket, String measurement, List<String> items, @NotBlank String type);
+
+    List<Item> queryHistory(String bucket, String measurement, Date startTime, Date endTime, List<String> items, @NotBlank String type);
 }

industry-system/cqcy-ei-influxdb/src/main/java/com/cqcy/ei/influxdb/service/impl/InFluxDBServiceImpl.java

@@ -1,20 +1,36 @@
 package com.cqcy.ei.influxdb.service.impl;
 
+import cn.hutool.core.collection.CollUtil;
 import cn.hutool.core.util.StrUtil;
 import com.cqcy.ei.influxdb.config.InfluxDBProperties;
+import com.cqcy.ei.influxdb.entity.Item;
 import com.cqcy.ei.influxdb.entity.MoveMeasurementEntity;
 import com.cqcy.ei.influxdb.entity.RangeEnum;
 import com.cqcy.ei.influxdb.mapper.StoreMapper;
 import com.cqcy.ei.influxdb.service.InFluxDBService;
 import com.cqcy.ei.influxdb.util.ClientUtil;
+import com.cqcy.ei.influxdb.util.InfluxDBUtil;
+import com.example.opc_common.entity.ItemParent;
+import com.example.opc_common.util.ConstantStr;
+import com.example.opc_common.util.DateUtil;
+import com.influxdb.client.InfluxDBClient;
+import com.influxdb.client.InfluxDBClientFactory;
+import com.influxdb.client.InfluxDBClientOptions;
 import com.influxdb.client.domain.Bucket;
 import com.influxdb.client.domain.Organization;
+import com.influxdb.client.domain.WritePrecision;
+import com.influxdb.client.write.Point;
+import com.influxdb.query.FluxTable;
 import lombok.AllArgsConstructor;
+import okhttp3.OkHttpClient;
 import org.springframework.stereotype.Service;
 
 import java.time.OffsetDateTime;
 import java.time.ZoneOffset;
+import java.util.ArrayList;
 import java.util.Date;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
 
 @Service
 @AllArgsConstructor
@@ -87,20 +103,20 @@ public class InFluxDBServiceImpl implements InFluxDBService {
     @Override
     public boolean deleteMeasurementByTime(Integer time, RangeEnum rangeEnum, String measurement, String bucket) {
         Date startTime = new Date();
-        Date endTime = rangeEnum.getEndTime(startTime,time);
+        Date endTime = rangeEnum.getEndTime(startTime, time);
         return deleteMeasurementByTime(endTime, startTime, measurement, bucket);
     }
 
     @Override
     public boolean moveMeasurementByTime(Date startTime, Date endTime, String bucket, String measurement, String targetBucket, String targetMeasurement) {
         return storeMapper.subMeterMainByDate(
-            new MoveMeasurementEntity()
-                .startTime(startTime)
-                .endTime(endTime)
-                .bucket(bucket)
-                .measurement(measurement)
-                .targetBucket(targetBucket)
-                .targetMeasurement(targetMeasurement)
+                new MoveMeasurementEntity()
+                        .startTime(startTime)
+                        .endTime(endTime)
+                        .bucket(bucket)
+                        .measurement(measurement)
+                        .targetBucket(targetBucket)
+                        .targetMeasurement(targetMeasurement)
         );
     }
 
@@ -126,7 +142,7 @@ public class InFluxDBServiceImpl implements InFluxDBService {
 
     @Override
     public boolean moveMeasurementByTime(Date startTime, Integer time, RangeEnum rangeEnum, String bucket, String measurement, String targetBucket, String targetMeasurement) {
-        Date endTime = rangeEnum.getEndTime(startTime,time);
+        Date endTime = rangeEnum.getEndTime(startTime, time);
         return moveMeasurementByTime(endTime, startTime, bucket, measurement, targetBucket, targetMeasurement);
     }
 
@@ -134,4 +150,140 @@ public class InFluxDBServiceImpl implements InFluxDBService {
     public Integer getMeasurementCountByTime(Date startTime, Date endTime) {
         return storeMapper.getMeasurementCountByTime(startTime.toInstant().getEpochSecond(), endTime.toInstant().getEpochSecond());
     }
+
+    @Override
+    public Boolean batchInsert(String bucket, String measurement, List<? extends ItemParent> itemParentList) {
+        OkHttpClient.Builder okHttpClientBuilder = new OkHttpClient.Builder()
+                .readTimeout(influxDBProperties.getReadTimeout(), TimeUnit.SECONDS)
+                .writeTimeout(influxDBProperties.getWriteTimeout(), TimeUnit.SECONDS)
+                .connectTimeout(influxDBProperties.getConnectTimeout(), TimeUnit.SECONDS);
+        // 设置客户端信息
+        InfluxDBClientOptions options = InfluxDBClientOptions.builder()
+                .okHttpClient(okHttpClientBuilder)
+                .bucket(bucket)
+                .url(influxDBProperties.getUrl())
+                .authenticateToken(influxDBProperties.getToken().toCharArray())
+                .org(influxDBProperties.getOrg())
+                .build();
+        // 创建连接配置对象
+        InfluxDBClient influxDBClient = InfluxDBClientFactory.create(options);
+        if (!checkBucketExist(bucket)) {
+            createBucket(bucket);
+        }
+        try {
+            // 构造要写入的Point集合
+            List<Point> points = new ArrayList<Point>();
+            itemParentList.forEach(itemParent -> {
+                points.add(Point.measurement(measurement)
+                        .addTag("type", ConstantStr.ORIGINAL_VALUE + "")
+                        .addField(itemParent.getItemReadName(), itemParent.getDataValue())
+                        .time(DateUtil.strChangeDate(itemParent.getDataTime(),
+                                "yyyy-MM-dd HH:mm:ss.SSS").getTime(), WritePrecision.MS));
+                points.add(Point.measurement(measurement)
+                        .addTag("type", ConstantStr.CALCULATED_VALUE + "")
+                        .addField(itemParent.getItemReadName(), itemParent.getCountDataValue())
+                        .time(DateUtil.strChangeDate(itemParent.getDataTime(),
+                                "yyyy-MM-dd HH:mm:ss.SSS").getTime(), WritePrecision.MS));
+            });
+            influxDBClient.getWriteApiBlocking().writePoints(points);
+        } finally {
+            influxDBClient.close();
+        }
+        return true;
+    }
+
+    @Override
+    public List<Item> queryLast(String bucket, String measurement, List<String> items, String type) {
+        OkHttpClient.Builder okHttpClientBuilder = new OkHttpClient.Builder()
+                .readTimeout(influxDBProperties.getReadTimeout(), TimeUnit.SECONDS)
+                .writeTimeout(influxDBProperties.getWriteTimeout(), TimeUnit.SECONDS)
+                .connectTimeout(influxDBProperties.getConnectTimeout(), TimeUnit.SECONDS);
+        // 设置客户端信息
+        InfluxDBClientOptions options = InfluxDBClientOptions.builder()
+                .okHttpClient(okHttpClientBuilder)
+                .bucket(bucket)
+                .url(influxDBProperties.getUrl())
+                .authenticateToken(influxDBProperties.getToken().toCharArray())
+                .org(influxDBProperties.getOrg())
+                .build();
+        // 创建连接配置对象
+        InfluxDBClient influxDBClient = InfluxDBClientFactory.create(options);
+        try {
+            // 构造要查询的sql
+            StringBuilder sql = new StringBuilder();
+            Date endTime = new Date();
+            Date startTime = RangeEnum.HOUR.getEndTime(endTime, 1);
+            //from 指定数据源bucket
+            sql.append("from(bucket: \"" + bucket + "\")")
+                    //|> 管道连接符
+                    //range 指定起始时间段
+                    .append("|> range(start: " + startTime.getTime() + ", stop: " + endTime.getTime() + ")")
+                    .append("|> filter(fn: (r) => r[\"type\"] == \"" + type + "\")")
+                    //filter 过滤
+                    .append("|> filter(fn: (r) => r[\"_measurement\"] == \"" + measurement + "\")");
+            if (CollUtil.isNotEmpty(items)) {
+                sql.append("|> filter(fn: (r) => ");
+                for (int i = 0; i < items.size(); i++) {
+                    sql.append("r[\"_field\"] ==\"" + items.get(i) + "\"");
+                    if (i != items.size() - 1) sql.append(" or ");
+
+                }
+                sql.append(")");
+            }
+            //排序
+            sql.append("|> sort(columns:[\"_field\",\"_time\"])")
+                    .append("|> last()");
+            System.out.println(sql);
+            List<FluxTable> fluxTableList = ClientUtil.client().getQueryApi().query(sql.toString());
+            return InfluxDBUtil.parseList(fluxTableList);
+        } finally {
+            influxDBClient.close();
+        }
+    }
+
+    @Override
+    public List<Item> queryHistory(String bucket, String measurement, Date startTime, Date endTime, List<String> items, String type) {
+        OkHttpClient.Builder okHttpClientBuilder = new OkHttpClient.Builder()
+                .readTimeout(influxDBProperties.getReadTimeout(), TimeUnit.SECONDS)
+                .writeTimeout(influxDBProperties.getWriteTimeout(), TimeUnit.SECONDS)
+                .connectTimeout(influxDBProperties.getConnectTimeout(), TimeUnit.SECONDS);
+        // 设置客户端信息
+        InfluxDBClientOptions options = InfluxDBClientOptions.builder()
+                .okHttpClient(okHttpClientBuilder)
+                .bucket(bucket)
+                .url(influxDBProperties.getUrl())
+                .authenticateToken(influxDBProperties.getToken().toCharArray())
+                .org(influxDBProperties.getOrg())
+                .build();
+        // 创建连接配置对象
+        InfluxDBClient influxDBClient = InfluxDBClientFactory.create(options);
+        try {
+            // 构造要查询的sql
+            StringBuilder sql = new StringBuilder();
+            //from 指定数据源bucket
+            sql.append("from(bucket: \"" + bucket + "\")")
+                    //|> 管道连接符
+                    //range 指定起始时间段
+                    .append("|> range(start: " + startTime.getTime() + ", stop: " + endTime.getTime() + ")")
+                    .append("|> filter(fn: (r) => r[\"type\"] == \"" + type + "\")")
+                    //filter 过滤
+                    .append("|> filter(fn: (r) => r[\"_measurement\"] == \"" + measurement + "\")");
+            if (CollUtil.isNotEmpty(items)) {
+                sql.append("|> filter(fn: (r) => ");
+                for (int i = 0; i < items.size(); i++) {
+                    sql.append("r[\"_field\"] ==\"" + items.get(i) + "\"");
+                    if (i != items.size() - 1) sql.append(" or ");
+
+                }
+                sql.append(")");
+            }
+            //排序
+            sql.append("|> sort(columns:[\"_field\",\"_time\"])");
+            System.out.println(sql);
+            List<FluxTable> fluxTableList = ClientUtil.client().getQueryApi().query(sql.toString());
+            return InfluxDBUtil.parseList(fluxTableList);
+        } finally {
+            influxDBClient.close();
+        }
+    }
 }

industry-system/cqcy-ei-influxdb/src/main/java/com/cqcy/ei/influxdb/util/InfluxDBUtil.java

@@ -43,13 +43,13 @@ public class InfluxDBUtil {
             return data;
         }
         Item item;
-        for (FluxTable fluxTable: list) {
+        for (FluxTable fluxTable : list) {
             List<FluxRecord> records = fluxTable.getRecords();
-            for (FluxRecord record: records) {
+            for (FluxRecord record : records) {
                 item = new Item()
-                    .name(record.getField())
-                    .value(Convert.toStr(record.getValue(), ""))
-                    .time(record.getTime());
+                        .name(record.getField())
+                        .value(Convert.toStr(record.getValue(), ""))
+                        .time(record.getTime());
                 data.add(item);
             }
         }
@@ -58,10 +58,11 @@ public class InfluxDBUtil {
 
     /**
      * 根据时间获取分表关系
-     * @param startTime 开始时间
-     * @param endTime 结束时间
+     *
+     * @param startTime  开始时间
+     * @param endTime    结束时间
      * @param properties client原数据
-     * @return List<Map<String, Object>>
+     * @return List<Map < String, Object>>
      */
     public static List<Map<String, Object>> getMeasurement(Date startTime, Date endTime, InfluxDBProperties properties) {
         List<Map<String, Object>> list = new ArrayList<>();
@@ -89,8 +90,8 @@ public class InfluxDBUtil {
                 int startMon = 0;
                 // 按年分库按月分表
                 for (int i = startYear; i <= endYear; i++) {
-                    endMon = i == endYear ? endMonth: 12;
-                    startMon = i == startYear ? startMonth: 1;
+                    endMon = i == endYear ? endMonth : 12;
+                    startMon = i == startYear ? startMonth : 1;
                     measurements = new ArrayList<>();
                     params = new HashMap<>();
                     params.put("bucket", properties.getBucket() + "_" + i);

industry-system/industry-da/src/main/java/com/example/opc_da/controller/ReportDataPolicyController.java

@@ -124,4 +124,28 @@ public class ReportDataPolicyController {
     public Result updatePolicyItemList(@RequestBody ReportDataPolicy reportDataPolicy) {
         return reportDataPolicyService.updatePolicyItemList(reportDataPolicy);
     }
+
+    /**
+     * 运行报表数据策略
+     *
+     * @param id
+     * @return
+     */
+    @RequestMapping(value = "/runReportDataPolicyById", method = RequestMethod.POST)
+    @WebLog(ServerEnum = ServerEnum.CLIENT, ModelEnum = ModelEnum.REPORTDATAPOLICY, OperationEnum = OperationEnum.RUN)
+    public Result runReportDataPolicyById(@RequestParam Integer id) {
+        return reportDataPolicyService.runReportDataPolicyById(id);
+    }
+
+    /**
+     * 停止报表数据策略
+     *
+     * @param id
+     * @return
+     */
+    @RequestMapping(value = "/stopReportDataPolicyById", method = RequestMethod.POST)
+    @WebLog(ServerEnum = ServerEnum.CLIENT, ModelEnum = ModelEnum.REPORTDATAPOLICY, OperationEnum = OperationEnum.STOPIT)
+    public Result stopReportDataPolicyById(@RequestParam Integer id) {
+        return reportDataPolicyService.stopReportDataPolicyById(id);
+    }
 }

industry-system/industry-da/src/main/java/com/example/opc_da/dao/ItemGroupDao.java

@@ -62,8 +62,6 @@ public interface ItemGroupDao {
 
     Integer updateItemByName(Integer itemGroupId, Item item);
 
-    List<Item> getAttributesItemList(Integer itemGroupId, Integer itemType);
-
     List<ItemGroup> getAllItemGroup1(String userId, Integer readMode);
 
     List<ItemGroup> getIgListByDsId(List<DataSource> dataSourceList);

industry-system/industry-da/src/main/java/com/example/opc_da/dao/ReportDataPolicyDao.java

@@ -119,4 +119,6 @@ public interface ReportDataPolicyDao {
     Integer delRdpItemByPIdDed(Integer reportDataPolicyId, List<ReportDataPolicyItem> policyItemList);
 
     List<ReportDataPolicyItem> getPublicItemList(Integer reportDataPolicyId, List<ReportDataPolicyItem> policyItemList);
+
+    Integer runReportDataPolicyById(Integer id, Integer runState, String cronId);
 }

industry-system/industry-da/src/main/java/com/example/opc_da/quartz/task/ReportDataPolicyTask.java

@@ -0,0 +1,130 @@
+package com.example.opc_da.quartz.task;
+
+import com.cqcy.ei.influxdb.service.InFluxDBService;
+import com.cqcy.ei.influxdb.service.QueryService;
+import com.example.opc_common.entity.ReportDataPolicy;
+import com.example.opc_common.entity.ReportDataPolicyItem;
+import com.example.opc_common.util.Blank;
+import com.example.opc_common.util.ConstantStr;
+import com.example.opc_common.util.DateUtil;
+import com.example.opc_da.dao.ItemGroupDao;
+import com.example.opc_da.dao.ReportDataPolicyDao;
+import com.example.opc_da.dao.ReportTableDao;
+import com.example.opc_da.quartz.QuartzService;
+import com.example.opc_da.task.AsyncTask;
+import com.example.opc_da.task.PolicyReadTask;
+import com.example.opc_da.task.ReportTableTask;
+import com.example.opc_da.util.QueryServiceUtil;
+import com.example.opc_da.util.RedisUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.quartz.JobDataMap;
+import org.quartz.JobExecutionContext;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.scheduling.quartz.QuartzJobBean;
+import org.springframework.stereotype.Component;
+
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.Timer;
+
+@Slf4j
+@Component
+public class ReportDataPolicyTask extends QuartzJobBean {
+
+    @Autowired
+    private ItemGroupDao itemGroupDao;
+
+    @Autowired
+    private ReportTableDao reportTableDao;
+
+    @Autowired
+    private QuartzService quartzService;
+
+    @Autowired
+    private RedisUtil redisUtil;
+
+
+    @Value("${opc_storage.time_format}")
+    private String timeFormat;
+
+    @Autowired
+    private ReportTableTask reportTableTask;
+
+    @Autowired
+    private QueryServiceUtil queryServiceUtil;
+
+    @Autowired
+    private ReportDataPolicyDao reportDataPolicyDao;
+
+    @Autowired
+    private QueryService queryService;
+
+    @Autowired
+    private InFluxDBService inFluxDBService;
+
+    @Value("${influxdb.bucket}")
+    private String bucket;
+
+    @Autowired
+    private AsyncTask asyncTask;
+
+
+    @Override
+    protected void executeInternal(JobExecutionContext context) {
+        JobDataMap jobDataMap = context.getJobDetail().getJobDataMap();
+        ReportDataPolicy reportDataPolicy = (ReportDataPolicy) jobDataMap.get("reportDataPolicy");
+        String cronId = jobDataMap.getString("cronId");
+
+        reportDataPolicyTask(reportDataPolicy, cronId);
+    }
+
+    /**
+     * 报表数据策略定时启动，执行的任务
+     *
+     * @param reportDataPolicy
+     * @param cronId
+     */
+    public void reportDataPolicyTask(ReportDataPolicy reportDataPolicy, String cronId) {
+        log.error("报表数据策略开始运行,{},执行开始，执行时间为{}", reportDataPolicy,
+                DateUtil.dateChangeStrYmdhmss(new Date()));
+        Integer id = reportDataPolicy.getId();
+        Integer readMode = reportDataPolicy.getReadMode();
+        Boolean flag = (Boolean) redisUtil.get(ConstantStr.REPORT_DATA_POLICY + id);
+        if (Blank.isEmpty(flag)) {
+            flag = false;
+        }
+        if (!flag) {
+            redisUtil.del(ConstantStr.REPORT_DATA_POLICY + id);
+            quartzService.deleteTask(cronId);
+            return;
+        }
+        //读取数据并保存
+        Timer timer = new Timer();
+        //生成Timer当天的开始、结束执行时间
+        String startTime = DateUtil.getCurrentYmd() + " " + reportDataPolicy.getStartReadTime();
+        String endTime = DateUtil.getCurrentYmd() + " " + reportDataPolicy.getEndReadTime();
+        Date startDate = DateUtil.strYmdhmsChangeDate(startTime);
+        Date endDate = DateUtil.strYmdhmsChangeDate(endTime);
+        //获取报表数据策略的所有数据项信息
+        List<ReportDataPolicyItem> policyItems = (List<ReportDataPolicyItem>) queryServiceUtil.exchangeData(
+                reportDataPolicyDao.getRdpItemByPolicyId(id)
+        );
+        reportDataPolicy.setPolicyItemList(policyItems);
+
+        //因为数据项来源同一个数据组，所以肯定是同一数据源
+        List<String> items = new ArrayList<>();
+        policyItems.forEach(item -> {
+            items.add((item.getItemReadName()));
+        });
+
+        PolicyReadTask policyReadTask = new PolicyReadTask(
+                timer, queryService, redisUtil, asyncTask, endDate.getTime(),
+                reportDataPolicy, policyItems.get(0).getDataSourceId().toString(), items
+        );
+        timer.scheduleAtFixedRate(policyReadTask,
+                startDate.getTime() > new Date().getTime() ? startDate : new Date(),
+                readMode.equals(ConstantStr.ON_FREQUENCY) ? (long) (1000 * reportDataPolicy.getModeValue()) : 1000);
+    }
+}

industry-system/industry-da/src/main/java/com/example/opc_da/service/ReportDataPolicyService.java

@@ -24,4 +24,8 @@ public interface ReportDataPolicyService {
     Result delPolicyItemByIdList(List<Integer> idList);
 
     Result updatePolicyItemList(ReportDataPolicy reportDataPolicy);
+
+    Result runReportDataPolicyById(Integer id);
+
+    Result stopReportDataPolicyById(Integer id);
 }

industry-system/industry-da/src/main/java/com/example/opc_da/service/impl/ReportDataPolicyServiceImpl.java

@@ -1,23 +1,28 @@
 package com.example.opc_da.service.impl;
 
+import com.cqcy.ei.influxdb.service.InFluxDBService;
 import com.example.opc_common.entity.ReportDataPolicy;
 import com.example.opc_common.entity.ReportDataPolicyItem;
 import com.example.opc_common.enums.ResultEnum;
 import com.example.opc_common.exception.CustomException;
 import com.example.opc_common.util.Blank;
 import com.example.opc_common.util.ConstantStr;
+import com.example.opc_common.util.DateUtil;
 import com.example.opc_common.util.Result;
 import com.example.opc_da.dao.ReportDataPolicyDao;
 import com.example.opc_da.dao.ReportTableDao;
-import com.example.opc_da.service.ItemGroupService;
+import com.example.opc_da.quartz.QuartzBean;
+import com.example.opc_da.quartz.QuartzService;
+import com.example.opc_da.quartz.task.ReportDataPolicyTask;
 import com.example.opc_da.service.ReportDataPolicyService;
 import com.example.opc_da.util.QueryServiceUtil;
+import com.example.opc_da.util.RedisUtil;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
-import java.util.ArrayList;
-import java.util.List;
+import java.util.*;
 
 @Service
 @Transactional
@@ -27,14 +32,26 @@ public class ReportDataPolicyServiceImpl implements ReportDataPolicyService {
     private ReportDataPolicyDao reportDataPolicyDao;
 
     @Autowired
-    private ItemGroupService itemGroupService;
+    private QueryServiceUtil queryServiceUtil;
 
     @Autowired
-    private QueryServiceUtil queryServiceUtil;
+    private InFluxDBService inFluxDBService;
+
+    @Value("${influxdb.bucket}")
+    private String bucket;
 
     @Autowired
     private ReportTableDao reportTableDao;
 
+    @Autowired
+    private RedisUtil redisUtil;
+
+    @Autowired
+    private QuartzService quartzService;
+
+    @Autowired
+    private ReportDataPolicyTask reportDataPolicyTask;
+
     @Override
     public synchronized Result addReportDataPolicy(ReportDataPolicy reportDataPolicy) {
         ReportDataPolicy oldReportDataPolicy = reportDataPolicyDao.getReportDataPolicyByName(reportDataPolicy.getReportDataPolicyName());
@@ -275,4 +292,48 @@ public class ReportDataPolicyServiceImpl implements ReportDataPolicyService {
         }
         return Result.ok("修改报表数据策略数据项成功");
     }
+
+    @Override
+    public Result runReportDataPolicyById(Integer id) {
+        ReportDataPolicy reportDataPolicy = reportDataPolicyDao.getReportDataPolicyById(id);
+        if (reportDataPolicy.getRunState() == ConstantStr.START_UP) {
+            return Result.ok("此报表数据策略已处于运行中");
+        }
+        redisUtil.set(ConstantStr.REPORT_DATA_POLICY + id, true);
+        //新增quarz定时器任务
+        String cronId = Blank.isEmpty(reportDataPolicy.getCronId()) ?
+                UUID.randomUUID().toString().replace("-", "") :
+                reportDataPolicy.getCronId();
+        String cron = "0 0 0 ? * ";
+        String readWeek = reportDataPolicy.getReadWeek();
+        if (Blank.isEmpty(readWeek)) {
+            cron = cron + "MON,TUE,WED,THU,FRI,SAT,SUN";
+            //立即执行一次定时器任务
+            reportDataPolicyTask.reportDataPolicyTask(reportDataPolicy, cronId);
+        } else {
+            cron = cron + readWeek;
+            for (String str : readWeek.split(",")) {
+                if (str.equals(DateUtil.getCurrentWeekEn())) {
+                    //立即执行一次Timer
+                    reportDataPolicyTask.reportDataPolicyTask(reportDataPolicy, cronId);
+                    break;
+                }
+            }
+        }
+        Map<String, Object> map = new HashMap<>();
+        map.put("reportDataPolicy", reportDataPolicy);
+        map.put("cronId", cronId);
+        quartzService.addTask(new QuartzBean(cronId, cronId, ReportDataPolicyTask.class, 0, cron, map));
+        reportDataPolicyDao.runReportDataPolicyById(id, ConstantStr.START_UP, cronId);
+        return Result.ok("启动成功");
+    }
+
+    @Override
+    public Result stopReportDataPolicyById(Integer id) {
+        ReportDataPolicy reportDataPolicy = reportDataPolicyDao.getReportDataPolicyById(id);
+        redisUtil.set(ConstantStr.REPORT_DATA_POLICY + id, false);
+        reportDataPolicyDao.runReportDataPolicyById(id, ConstantStr.STOP_IT, null);
+        quartzService.deleteTask(reportDataPolicy.getCronId());
+        return Result.ok("停用成功");
+    }
 }

industry-system/industry-da/src/main/java/com/example/opc_da/task/AsyncTask.java

@@ -0,0 +1,30 @@
+package com.example.opc_da.task;
+
+import com.cqcy.ei.influxdb.service.InFluxDBService;
+import com.example.opc_common.entity.ItemParent;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+
+/**
+ * Java异步任务
+ */
+@Component
+@Async("threadPoolTaskExecutor")
+@Slf4j
+public class AsyncTask {
+
+    @Autowired
+    private InFluxDBService inFluxDBService;
+
+    @Value("${influxdb.bucket}")
+    private String bucket;
+
+    public void addData(Integer reportDataPolicyId, List<? extends ItemParent> itemParentList) {
+        inFluxDBService.batchInsert(bucket, reportDataPolicyId.toString(), itemParentList);
+    }
+}

industry-system/industry-da/src/main/java/com/example/opc_da/task/PolicyReadTask.java

@@ -0,0 +1,190 @@
+package com.example.opc_da.task;
+
+import com.cqcy.ei.influxdb.entity.Item;
+import com.cqcy.ei.influxdb.service.QueryService;
+import com.example.opc_common.entity.ReportDataPolicy;
+import com.example.opc_common.entity.ReportDataPolicyItem;
+import com.example.opc_common.util.Blank;
+import com.example.opc_common.util.ConstantStr;
+import com.example.opc_common.util.MathUtil;
+import com.example.opc_da.util.QueryServiceUtil;
+import com.example.opc_da.util.RedisUtil;
+import lombok.extern.slf4j.Slf4j;
+
+import java.math.BigDecimal;
+import java.util.*;
+import java.util.stream.Collectors;
+
+@Slf4j
+public class PolicyReadTask extends TimerTask {
+    private final Timer timer;
+    private final QueryService queryService;
+    private final RedisUtil redisUtil;
+    private final AsyncTask asyncTask;
+    private final Long endTime;
+    private final ReportDataPolicy reportDataPolicy;
+    private final String dataSourceId;
+    private final List<String> items;
+
+    /**
+     * 有参构造
+     */
+    /**
+     * @param timer        任务停止时，摧毁timer
+     * @param queryService 从influxdb获取数据
+     * @param redisUtil    往缓存中添加获取数据
+     * @param asyncTask    使用异步任务，将获取的数据添加到数据库
+     * @param endTime      定时任务停止时间
+     */
+    public PolicyReadTask(Timer timer,
+                          QueryService queryService,
+                          RedisUtil redisUtil,
+                          AsyncTask asyncTask,
+                          Long endTime,
+                          ReportDataPolicy reportDataPolicy,
+                          String dataSourceId,
+                          List<String> items) {
+        this.timer = timer;
+        this.queryService = queryService;
+        this.redisUtil = redisUtil;
+        this.asyncTask = asyncTask;
+        this.endTime = endTime;
+        this.reportDataPolicy = reportDataPolicy;
+        this.dataSourceId = dataSourceId;
+        this.items = items;
+    }
+
+    @Override
+    public void run() {
+        log.info("定时器内部开始执行");
+        Integer readMode = reportDataPolicy.getReadMode();
+        Integer id = reportDataPolicy.getId();
+        List<ReportDataPolicyItem> policyItemList = reportDataPolicy.getPolicyItemList();
+        //查询数据项相关的数据
+        List<Item> itemDataByLast = queryService.getItemDataByLast(items, dataSourceId);
+        //判断数据是否为重复数据
+        if (true) {
+            return;
+        }
+        //将得到的数据转换为Map<数据项名称，Item实体类>
+        Map<String, Item> map = new HashMap<>();
+        if (Blank.isNotEmpty(itemDataByLast)) {
+            for (Item item : itemDataByLast) {
+                map.put(item.getName(), item);
+            }
+        }
+        //将报表数据策略的数据项，赋值相应的值
+        for (int i = 0; i < policyItemList.size(); i++) {
+            ReportDataPolicyItem policyItem = policyItemList.get(i);
+            Item item = map.get(policyItem.getItemReadName());
+            QueryServiceUtil.itemParentCountValue(policyItem, item);
+        }
+
+        //按照频率读取
+        if (readMode.equals(ConstantStr.ON_FREQUENCY)) {
+            asyncTask.addData(id, policyItemList);
+            //按照值读取
+        } else if (readMode.equals(ConstantStr.ON_CHANGE)) {
+            changeRead(id, policyItemList);
+            //按照条件读取
+        } else if (readMode.equals(ConstantStr.EXCEED_SET_VALUE)) {
+            eventRead(id, policyItemList);
+        }
+
+        if (!getPolicyRunState(reportDataPolicy.getId())) {
+            timer.cancel();
+        }
+        //判断数据组是否到达了，今天的结束时间
+        if (System.currentTimeMillis() >= endTime) {
+            timer.cancel();
+        }
+    }
+
+    /**
+     * 按照值改变读取
+     *
+     * @param id
+     * @param policyItemList
+     */
+    private void changeRead(Integer id, List<ReportDataPolicyItem> policyItemList) {
+        List<ReportDataPolicyItem> validPolicyItemList = new ArrayList<>();
+        for (ReportDataPolicyItem policyItem : policyItemList) {
+            Integer itemId = policyItem.getItemId();
+            BigDecimal dataValue = new BigDecimal(policyItem.getDataValue());
+            if (MathUtil.isMeetChange(
+                    (BigDecimal) redisUtil.get(ConstantStr.VALUE + itemId)
+                    , dataValue
+                    , new BigDecimal(reportDataPolicy.getModeValue())
+                    , reportDataPolicy.getReadModeType()
+            )) validPolicyItemList.add(policyItem);
+            redisUtil.set(ConstantStr.VALUE + itemId, dataValue);
+        }
+        if (Blank.isNotEmpty(validPolicyItemList)) asyncTask.addData(id, validPolicyItemList);
+    }
+
+    /**
+     * 条件读取
+     *
+     * @param id
+     * @param policyItemList
+     */
+    private void eventRead(Integer id, List<ReportDataPolicyItem> policyItemList) {
+        //条件数据项（事件驱动数据项）
+        Integer eventItemId = reportDataPolicy.getEventItemId();
+        //条件模式（事件驱动模式，0大于，1小于(0,1,值只对应数值类型)，2等于，3动态大于（只记录超过设定值那一瞬间的数据），4动态小于（只记录低于设定值那一瞬间的数据））
+        Integer eventMode = reportDataPolicy.getEventMode();
+        //模式值
+        BigDecimal modeValue = new BigDecimal(reportDataPolicy.getModeValue());
+        //过滤得到事件驱动数据项信息
+        List<ReportDataPolicyItem> collect = policyItemList.stream()
+                .filter(policyItem -> policyItem.getItemId().equals(eventItemId))
+                .collect(Collectors.toList());
+        //判断事件驱动项是否符合事件驱动条件
+        ReportDataPolicyItem eventPolicyItem = collect.get(0);
+        BigDecimal dataValue = new BigDecimal(eventPolicyItem.getDataValue());
+        Boolean eventFlag = false;
+        //如果条件模式是，大于
+        if (eventMode.equals(ConstantStr.EVENT_MODEL_EXCEED)) {
+            if (dataValue.compareTo(modeValue) == 1) eventFlag = true;
+            //如果条件模式是，小于
+        } else if (eventMode.equals(ConstantStr.EVENT_MODEL_LOWER)) {
+            if (dataValue.compareTo(modeValue) == -1) eventFlag = true;
+            //如果条件模式是，等于
+        } else if (eventMode.equals(ConstantStr.EVENT_MODEL_EQUAL)) {
+            if (dataValue.compareTo(modeValue) == 0) eventFlag = true;
+            //如果条件模式是，动态大于
+        } else if (eventMode.equals(ConstantStr.EVENT_TRENDS_EXCEED)) {
+            if (((BigDecimal) redisUtil.get(ConstantStr.VALUE + eventItemId)).compareTo(modeValue) != 1) {
+                if (dataValue.compareTo(modeValue) == 1) eventFlag = true;
+            }
+            redisUtil.set(ConstantStr.VALUE + eventItemId, dataValue);
+            //如果条件模式是，动态小于
+        } else if (eventMode.equals(ConstantStr.EVENT_TRENDS_LOWER)) {
+            if (((BigDecimal) redisUtil.get(ConstantStr.VALUE + eventItemId)).compareTo(modeValue) != -1) {
+                if (dataValue.compareTo(modeValue) == -1) eventFlag = true;
+            }
+            redisUtil.set(ConstantStr.VALUE + eventItemId, dataValue);
+        }
+        if (eventFlag) {
+            //异步生成事件驱动报表
+
+            //向indb添加数据
+            asyncTask.addData(id, policyItemList);
+        }
+    }
+
+    /**
+     * 获取redis中此时报表数据策略的运行状态
+     *
+     * @param id
+     * @return
+     */
+    public Boolean getPolicyRunState(Integer id) {
+        Boolean flage = (Boolean) redisUtil.get(ConstantStr.REPORT_DATA_POLICY + id);
+        if (Blank.isEmpty(flage)) {
+            flage = false;
+        }
+        return flage;
+    }
+
+}

industry-system/industry-da/src/main/java/com/example/opc_da/util/QueryServiceUtil.java

@@ -85,42 +85,7 @@ public class QueryServiceUtil {
             for (int i = 0; i < itemList.size(); i++) {
                 com.example.opc_common.entity.Item commonItem = itemList.get(i);
                 Item item = mapData.get(commonItem.getDataSourceId().toString()).get(commonItem.getItemReadName());
-                if (Blank.isNotEmpty(item)) {
-                    //给数据项赋值
-                    String value = item.getValue();
-                    //根据数据项模型，赋值相应的计算值
-                    if (Blank.isNotEmpty(commonItem.getDataModelId())) {
-                        if (commonItem.getModelType().equals(ConstantStr.VALUE_REPLACE)) {
-                            String operationRule = commonItem.getOperationRule();
-                            if (Blank.isNotEmpty(operationRule)) {
-                                if (operationRule.toLowerCase().equals(value.toLowerCase())) {
-                                    commonItem.setCountDataValue(commonItem.getReplacingValue());
-                                }
-                            }
-                            String operationRuleTwo = commonItem.getOperationRuleTwo();
-                            if (Blank.isNotEmpty(operationRuleTwo)) {
-                                if (operationRuleTwo.toLowerCase().equals(value.toLowerCase())) {
-                                    commonItem.setCountDataValue(commonItem.getReplacingValueTwo());
-                                }
-                            }
-                        } else {
-                            try {
-                                BigDecimal bigDecimal = JSON.parseObject(value, BigDecimal.class);
-                                commonItem.setCountDataValue(
-                                        MathUtil.quadricOperation(
-                                                commonItem.getMathParameter(), commonItem.getOperationRule(), bigDecimal
-                                        ).toString()
-                                );
-                            } catch (Exception e) {
-                                commonItem.setCountDataValue(value);
-                            }
-                        }
-                    } else {
-                        commonItem.setCountDataValue(value);
-                    }
-                    commonItem.setDataValue(value);
-                    commonItem.setDataTime(item.getTime());
-                }
+                itemParentCountValue(commonItem, item);
             }
 
             //将对应的数据项值赋值给相应的泛型值
@@ -130,6 +95,54 @@ public class QueryServiceUtil {
     }
 
     /**
+     * 将influxdb接口传来的item对象，赋值给泛型（且是ItemParent的子类），赋值原始值，计算值，时间
+     *
+     * @param itemParent
+     * @param item
+     * @param <T>
+     * @return
+     */
+    public static <T extends ItemParent> T itemParentCountValue(T itemParent, com.cqcy.ei.influxdb.entity.Item item) {
+        if (Blank.isNotEmpty(item)) {
+            //给数据项赋值
+            String value = item.getValue();
+            //根据数据项模型，赋值相应的计算值
+            if (Blank.isNotEmpty(itemParent.getDataModelId())) {
+                if (itemParent.getModelType().equals(ConstantStr.VALUE_REPLACE)) {
+                    String operationRule = itemParent.getOperationRule();
+                    if (Blank.isNotEmpty(operationRule)) {
+                        if (operationRule.toLowerCase().equals(value.toLowerCase())) {
+                            itemParent.setCountDataValue(itemParent.getReplacingValue());
+                        }
+                    }
+                    String operationRuleTwo = itemParent.getOperationRuleTwo();
+                    if (Blank.isNotEmpty(operationRuleTwo)) {
+                        if (operationRuleTwo.toLowerCase().equals(value.toLowerCase())) {
+                            itemParent.setCountDataValue(itemParent.getReplacingValueTwo());
+                        }
+                    }
+                } else {
+                    try {
+                        BigDecimal bigDecimal = JSON.parseObject(value, BigDecimal.class);
+                        itemParent.setCountDataValue(
+                                MathUtil.quadricOperation(
+                                        itemParent.getMathParameter(), itemParent.getOperationRule(), bigDecimal
+                                ).toString()
+                        );
+                    } catch (Exception e) {
+                        itemParent.setCountDataValue(value);
+                    }
+                }
+            } else {
+                itemParent.setCountDataValue(value);
+            }
+            itemParent.setDataValue(value);
+            itemParent.setDataTime(item.getTime());
+        }
+        return itemParent;
+    }
+
+    /**
      * 将传入进来的集合数据项，是ItemsParent的子类，赋值
      * 获取历史数据
      *
@@ -177,49 +190,7 @@ public class QueryServiceUtil {
             for (int i = 0; i < itemList.size(); i++) {
                 com.example.opc_common.entity.Item commonItem = itemList.get(i);
                 List<Item> items = mapData.get(commonItem.getDataSourceId().toString()).get(commonItem.getItemReadName());
-                if (Blank.isNotEmpty(items)) {
-                    List<String> countDataValueList = new ArrayList<>();
-                    List<String> dataValueList = new ArrayList<>();
-                    List<String> dataTimeList = new ArrayList<>();
-                    items.forEach(item -> {
-                        String value = item.getValue();
-                        //根据数据项模型，赋值相应的计算值
-                        if (Blank.isNotEmpty(commonItem.getDataModelId())) {
-                            if (commonItem.getModelType().equals(ConstantStr.VALUE_REPLACE)) {
-                                String operationRule = commonItem.getOperationRule();
-                                if (Blank.isNotEmpty(operationRule)) {
-                                    if (operationRule.toLowerCase().equals(value.toLowerCase())) {
-                                        countDataValueList.add(commonItem.getReplacingValue());
-                                    }
-                                }
-                                String operationRuleTwo = commonItem.getOperationRuleTwo();
-                                if (Blank.isNotEmpty(operationRuleTwo)) {
-                                    if (operationRuleTwo.toLowerCase().equals(value.toLowerCase())) {
-                                        countDataValueList.add(commonItem.getReplacingValueTwo());
-                                    }
-                                }
-                            } else {
-                                try {
-                                    BigDecimal bigDecimal = JSON.parseObject(value, BigDecimal.class);
-                                    countDataValueList.add(
-                                            MathUtil.quadricOperation(
-                                                    commonItem.getMathParameter(), commonItem.getOperationRule(), bigDecimal
-                                            ).toString()
-                                    );
-                                } catch (Exception e) {
-                                    countDataValueList.add(value);
-                                }
-                            }
-                        } else {
-                            countDataValueList.add(value);
-                        }
-                        dataValueList.add(item.getValue());
-                        dataTimeList.add(item.getTime());
-                    });
-                    commonItem.setCountDataValueList(countDataValueList);
-                    commonItem.setDataValueList(dataValueList);
-                    commonItem.setDataTimeList(dataTimeList);
-                }
+                itemsParentCountValue(commonItem, items);
             }
 
             //将对应的数据项值赋值给相应的泛型值
@@ -228,4 +199,59 @@ public class QueryServiceUtil {
         return tList;
     }
 
+    /**
+     * 将influxdb接口传来的item集合，赋值给泛型（且是ItemsParent的子类），赋值原始值，计算值，时间
+     *
+     * @param itemsParent
+     * @param items
+     * @param <T>
+     * @return
+     */
+    public static <T extends ItemsParent> T itemsParentCountValue(T itemsParent, List<com.cqcy.ei.influxdb.entity.Item> items) {
+        if (Blank.isNotEmpty(items)) {
+            List<String> countDataValueList = new ArrayList<>();
+            List<String> dataValueList = new ArrayList<>();
+            List<String> dataTimeList = new ArrayList<>();
+            items.forEach(item -> {
+                String value = item.getValue();
+                //根据数据项模型，赋值相应的计算值
+                if (Blank.isNotEmpty(itemsParent.getDataModelId())) {
+                    if (itemsParent.getModelType().equals(ConstantStr.VALUE_REPLACE)) {
+                        String operationRule = itemsParent.getOperationRule();
+                        if (Blank.isNotEmpty(operationRule)) {
+                            if (operationRule.toLowerCase().equals(value.toLowerCase())) {
+                                countDataValueList.add(itemsParent.getReplacingValue());
+                            }
+                        }
+                        String operationRuleTwo = itemsParent.getOperationRuleTwo();
+                        if (Blank.isNotEmpty(operationRuleTwo)) {
+                            if (operationRuleTwo.toLowerCase().equals(value.toLowerCase())) {
+                                countDataValueList.add(itemsParent.getReplacingValueTwo());
+                            }
+                        }
+                    } else {
+                        try {
+                            BigDecimal bigDecimal = JSON.parseObject(value, BigDecimal.class);
+                            countDataValueList.add(
+                                    MathUtil.quadricOperation(
+                                            itemsParent.getMathParameter(), itemsParent.getOperationRule(), bigDecimal
+                                    ).toString()
+                            );
+                        } catch (Exception e) {
+                            countDataValueList.add(value);
+                        }
+                    }
+                } else {
+                    countDataValueList.add(value);
+                }
+                dataValueList.add(item.getValue());
+                dataTimeList.add(item.getTime());
+            });
+            itemsParent.setCountDataValueList(countDataValueList);
+            itemsParent.setDataValueList(dataValueList);
+            itemsParent.setDataTimeList(dataTimeList);
+        }
+        return itemsParent;
+    }
+
 }

industry-system/industry-da/src/main/resources/application.yml

@@ -1,6 +1,6 @@
 server:
   max-http-header-size: 10MB
-  
+
 spring:
   profiles:
     active: @spring.profiles.active@
@@ -37,3 +37,6 @@ telegraf:
   client:
     port: 12080
 
+influxdb:
+  bucket: filter
+

industry-system/industry-da/src/main/resources/mapper/ItemGroupDao.xml

@@ -263,30 +263,6 @@
           and data_source_id = #{dataSourceId}
     </select>
 
-    <select id="getAttributesItemList" resultType="com.example.opc_common.entity.Item">
-        select ti.id,
-        ti.item_group_id,
-        ti.item_name,
-        ti.item_read_name,
-        ti.item_type,
-        ti.node_index,
-        ti.data_type,
-        ti.upper_limit,
-        ti.lower_limit,
-        ti.`describe`,
-        ti.unit,
-        ti.data_model_id,
-        ti.is_driver_item,
-        tdm.operation_rule
-        FROM t_item ti
-        LEFT JOIN t_data_model tdm ON ti.data_model_id = tdm.id
-        where ti.item_group_id = #{itemGroupId}
-        <if test="itemType != null">
-            and ti.item_type = #{itemType}
-        </if>
-        order by ti.is_driver_item DESC, ti.item_name
-    </select>
-
     <select id="getAllItemGroup1" resultType="com.example.opc_common.entity.ItemGroup">
         select
         <include refid="itemGroup"/>

industry-system/industry-da/src/main/resources/mapper/ReportDataPolicyDao.xml

@@ -17,7 +17,8 @@
             useGeneratedKeys="true"
             keyProperty="id">
         insert into t_report_data_policy(report_data_policy_name, policy_remake, item_group_id, read_mode, event_mode,
-                                         read_mode_type, mode_value, event_item_id, read_week, start_read_time, end_read_time,
+                                         read_mode_type, mode_value, event_item_id, read_week, start_read_time,
+                                         end_read_time,
                                          create_time)
             value (#{reportDataPolicyName},#{policyRemake},#{itemGroupId},#{readMode},#{eventMode},
             #{readModeType},#{modeValue},#{eventItemId},#{readWeek},#{startReadTime},#{endReadTime},now())
@@ -50,6 +51,15 @@
         where id = #{id}
     </update>
 
+    <update id="runReportDataPolicyById">
+        update t_report_data_policy
+        set run_state=#{runState}
+        <if test="cronId != null and cronId != ''">
+            , cron_id = #{cronId}
+        </if>
+        where id = #{id}
+    </update>
+
     <delete id="delRdpItemByPolicyId">
         delete
         from t_report_data_policy_item

report-client/src/api/reportDataPolicy.js

@@ -128,4 +128,36 @@ export function updatePolicyItemList(data) {
         method: 'post',
         data
     })
+}
+
+/**
+ * 运行报表数据策略
+ * @param data
+ * @returns {AxiosPromise}
+ */
+export function runReportDataPolicyById(data) {
+    return request({
+        url: '/reportDataPolicy/runReportDataPolicyById',
+        headers: {
+            isToken: true
+        },
+        method: 'post',
+        params:data
+    })
+}
+
+/**
+ * 停止报表数据策略
+ * @param data
+ * @returns {AxiosPromise}
+ */
+export function stopReportDataPolicyById(data) {
+    return request({
+        url: '/reportDataPolicy/stopReportDataPolicyById',
+        headers: {
+            isToken: true
+        },
+        method: 'post',
+        params:data
+    })
 }

report-client/src/components/HeaderMain/dataPolicyList.vue

@@ -32,10 +32,10 @@
         <!-- 数据组列表右键操作 -->
         <ul v-show="visibleChildMenu" :style="{ left: menuLeft + 'px', top: menuTop + 'px' }"
             class="contextmenu contextmenu-black">
-            <li v-if="this.choosePolicyData && this.choosePolicyData.runState == 0" @click="runOrStopGroup(1)">
+            <li v-if="this.choosePolicyData && this.choosePolicyData.runState == 0" @click="runReportDataPolicyById()">
                 运行
             </li>
-            <li v-if="this.choosePolicyData && this.choosePolicyData.runState != 0" @click="runOrStopGroup(0)">
+            <li v-if="this.choosePolicyData && this.choosePolicyData.runState != 0" @click="stopReportDataPolicyById()">
                 停止
             </li>
             <li @click="updateReportDataPolicy">修改</li>
@@ -50,7 +50,8 @@
 <script>
 import {
     getAllReportDataPolicy, updateReportDataPolicy,
-    getReportDataPolicyById, delReportDataPolicyById
+    getReportDataPolicyById, delReportDataPolicyById,
+    runReportDataPolicyById, stopReportDataPolicyById
 } from "@/api/reportDataPolicy"
 
 import {
@@ -113,32 +114,10 @@ export default {
                         showAlertWin(this, null, e);
                     resolve([]);
                 });
-            } else {
-                // getItemGroupById(node.data.id).then(res => {
-                //   if (!res.data) {
-                //     resolve([])
-                //     return
-                //   }
-                //   let arr = res.data.itemList
-                //   for (let i in arr) {
-                //     arr[i].leaf = true
-                //   }
-                //   resolve(arr)
-                // }).catch((e) => {`
-                //   resolve([])
-                //   showAlertWin(this, e)
-                // })
             }
         },
         /** 刷新数据组列表 */
         refreshGroupData() {
-            // if (this.chooseGroupParentNode) {
-            //   let node = this.chooseGroupParentNode
-            //   if (node) {
-            //     node.loaded = false
-            //     node.expand()
-            //   }
-            // }
             let node = this.$refs.groupTree.getNode(-1);
             if (node) {
                 node.loaded = false;
@@ -222,8 +201,26 @@ export default {
                 query: { id: this.choosePolicyData.id }
             })
 
+        },
+        /** 运行报表数据策略 */
+        runReportDataPolicyById() {
+            this.closeMenu()
+            runReportDataPolicyById({ id: this.choosePolicyData.id }).then(res => {
+                console.log(res)
+                // 刷新数据组数据
+                this.refreshGroupData();
+            })
+        },
+        /** 停止报表数据策略 */
+        stopReportDataPolicyById() {
+            this.closeMenu()
+            stopReportDataPolicyById({ id: this.choosePolicyData.id }).then(res => {
+                console.log(res)
+                // 刷新数据组数据
+                this.refreshGroupData();
+            })
         }
     },
 
 }
-</script>
+</script>