解决冲突

amos-boot-system-jxiop/amos-boot-module-jxiop-bigscreen-biz/src/main/java/com/yeejoin/amos/boot/module/jxiop/biz/config/TdEngineAnalysisConfig.java

+package com.yeejoin.amos.boot.module.jxiop.biz.config;
+
+
+import com.alibaba.druid.pool.DruidDataSource;
+import com.baomidou.mybatisplus.extension.spring.MybatisSqlSessionFactoryBean;
+import org.apache.ibatis.session.SqlSessionFactory;
+import org.mybatis.spring.annotation.MapperScan;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
+
+import javax.sql.DataSource;
+
+/**
+ * 从数据源配置
+ * 若需要配置更多数据源 , 直接在yml中添加数据源配置再增加相应的新的数据源配置类即可
+ */
+@Configuration
+@MapperScan(basePackages = "com.yeejoin.amos.boot.module.jxiop.biz.tdanalysismapper", sqlSessionFactoryRef = "taosAnalysisSqlSessionFactory")
+public class TdEngineAnalysisConfig {
+    private Logger logger = LoggerFactory.getLogger(TdEngineAnalysisConfig.class);
+    // 精确到 cluster 目录，以便跟其他数据源隔离
+    private static final String MAPPER_LOCATION = "classpath*:mapper/tdengineanalysis/*.xml";
+
+
+    @Value("${spring.db4.datasource.url}")
+    private String dbUrl;
+
+    @Value("${spring.db4.datasource.username}")
+    private String username;
+
+    @Value("${spring.db4.datasource.password}")
+    private String password;
+
+    @Value("${spring.db4.datasource.driver-class-name}")
+    private String driverClassName;
+
+
+    @Bean(name = "taosAnalysisDataSource")   //声明其为Bean实例
+    public DataSource clusterDataSource() {
+        DruidDataSource datasource = new DruidDataSource();
+
+        datasource.setUrl(this.dbUrl);
+        datasource.setUsername(username);
+        datasource.setPassword(password);
+        datasource.setDriverClassName(driverClassName);
+        return datasource;
+    }
+
+    @Bean(name = "taosAnalysisTransactionManager")
+    public DataSourceTransactionManager clusterTransactionManager() {
+        return new DataSourceTransactionManager(clusterDataSource());
+    }
+
+    @Bean(name = "taosAnalysisSqlSessionFactory")
+    public SqlSessionFactory clusterSqlSessionFactory(@Qualifier("taosDataSource") DataSource culsterDataSource)
+            throws Exception {
+        final MybatisSqlSessionFactoryBean sessionFactory = new MybatisSqlSessionFactoryBean();
+        sessionFactory.setDataSource(culsterDataSource);
+        sessionFactory.setMapperLocations(new PathMatchingResourcePatternResolver()
+                .getResources(TdEngineAnalysisConfig.MAPPER_LOCATION));
+        sessionFactory.setTypeAliasesPackage("com.yeejoin.amos.boot.module.jxiop.biz.entity");
+        //mybatis 数据库字段与实体类属性驼峰映射配置
+        sessionFactory.getObject().getConfiguration().setMapUnderscoreToCamelCase(true);
+        return sessionFactory.getObject();
+    }
+}
+

amos-boot-system-jxiop/amos-boot-module-jxiop-bigscreen-biz/src/main/java/com/yeejoin/amos/boot/module/jxiop/biz/constants/CommonConstans.java

@@ -130,9 +130,9 @@ public class CommonConstans {
     public static final String YEAR_GEN_ATTAINMENT_RATE = "年计划完成率";
     //年利用小时数
     public static final String YEAR_GEN_HOURS = "年利用小时数";
-    public static final String WIND_SPEED_THIRTY_SECONDS = "30秒平均风速";
-    public static final String TOTAL_RADIATION = "辐照度";
-    public static final String TOTAL_RADIATION_SUM = "累计辐照度";
+    public static final String WIND_SPEED_THIRTY_SECONDS = "总30秒平均风速";
+    public static final String TOTAL_RADIATION = "总辐照度";
+    public static final String TOTAL_RADIATION_SUM = "总累计辐照度";
     public static final String ACTIVE_POWER = "有功功率";
 
 

amos-boot-system-jxiop/amos-boot-module-jxiop-bigscreen-biz/src/main/java/com/yeejoin/amos/boot/module/jxiop/biz/entity/IndicatorData.java

@@ -2,6 +2,7 @@ package com.yeejoin.amos.boot.module.jxiop.biz.entity;
 
 import lombok.Data;
 
+import java.io.PipedReader;
 import java.util.Date;
 
 @Data
@@ -23,7 +24,7 @@ public class IndicatorData {
 
     private String  unit;
 
-    private String  value;
+    private String  value="0";
 
     private Float valueF ;
 
@@ -35,4 +36,11 @@ public class IndicatorData {
 
     private String  displayName;
     private String  xtime;
+    private String pointSeq;
+    private String pointAddress;
+    private String pointLocation;
+    private String pointType;
+    private String pointName;
+    private String dasTime;
+
 }

amos-boot-system-jxiop/amos-boot-module-jxiop-bigscreen-biz/src/main/java/com/yeejoin/amos/boot/module/jxiop/biz/service/MonitorService.java

@@ -89,7 +89,6 @@ public interface MonitorService {
     IPage<Map> getStationOverViewFanByStationId( String stationId);
     /**
      * @Descritpion 根据入参动态获取全国发电量数据
-     * @param stationId
      * @return
      */
     Map<String,Object> gettimedateyfd( );
@@ -109,5 +108,15 @@ public interface MonitorService {
      */
     ResultsData getElectricQuantityList(int current, int size, StationBasic stationBasic);
 
+    /**
+     * @Description 动态获取电量表计数据
+     */
     void getTotalData();
+
+    /**
+     * @Description 获取全国功率曲线
+     * @return
+     */
+    Map<String, Object> getDetailsWindSpeedAlldataqg();
+
 }

amos-boot-system-jxiop/amos-boot-module-jxiop-bigscreen-biz/src/main/java/com/yeejoin/amos/boot/module/jxiop/biz/service/impl/MonitorServiceImpl.java

@@ -19,8 +19,10 @@ import com.yeejoin.amos.boot.module.jxiop.api.util.HttpRequestUtil;
 import com.yeejoin.amos.boot.module.jxiop.biz.ESDto.ESMoonPowerGeneration;
 import com.yeejoin.amos.boot.module.jxiop.biz.constants.CommonConstans;
 import com.yeejoin.amos.boot.module.jxiop.biz.dto.*;
+import com.yeejoin.amos.boot.module.jxiop.biz.entity.IndicatorData;
 import com.yeejoin.amos.boot.module.jxiop.biz.service.CoreCommonService;
 import com.yeejoin.amos.boot.module.jxiop.biz.service.MonitorService;
+import com.yeejoin.amos.boot.module.jxiop.biz.tdanalysismapper.IndicatorDataNewMapper;
 import lombok.RequiredArgsConstructor;
 import org.apache.commons.io.IOUtils;
 import org.eclipse.paho.client.mqttv3.MqttException;
@@ -43,6 +45,7 @@ import java.text.DecimalFormat;
 import java.text.SimpleDateFormat;
 import java.time.LocalDate;
 import java.util.*;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 
 import static com.alibaba.fastjson.JSON.parseArray;
@@ -57,6 +60,7 @@ public class MonitorServiceImpl implements MonitorService {
     private final HttpRequestUtil httpRequestUtil;
     private final CoreCommonService coreCommonService;
     private final StationElectricityMeterMapper stationElectricityMeterMapper;
+    private final IndicatorDataNewMapper indicatorDataNewMapper;
     @Value("classpath:/json/overview.json")
     private Resource overview;
     @Value("classpath:/json/overviewGF.json")
@@ -893,4 +897,109 @@ public class MonitorServiceImpl implements MonitorService {
 
         }
     }
+
+
+
+    public  Map<String, Object> getDetailsWindSpeedAlldataqg() {
+
+        List<StationCacheInfoDto> stationCacheInfoDtos  = getListStationCacheInfoDto();
+        AtomicReference<Double> installedCapacity = new AtomicReference<>(0.0);
+        stationCacheInfoDtos.forEach( stationCacheInfoDto-> {
+            installedCapacity.updateAndGet(v -> v + Double.parseDouble(stationCacheInfoDto.getInstalledCapacity()));
+        });
+        Map<String, Object> map = new HashMap<>();
+        List<String> values = new ArrayList<>();
+
+        List<String> time = new ArrayList<>();
+        Map<String, Object> activePowerInfo = new HashMap<>();
+        List<IndicatorData> activePowerList = new ArrayList<>();
+        activePowerList = indicatorDataNewMapper.selectDataByequipmentIndexNameAndtimeqgNew(CommonConstans.ACTIVE_POWER);
+        Map<String,List<IndicatorData>> stringObjectMap =activePowerList.stream().collect(Collectors.groupingBy(IndicatorData::getDasTime));
+        stringObjectMap.keySet().stream().forEach(dasTime -> {
+            time.add(dasTime.split(":")[1]);
+            Double vl=  stringObjectMap.get(dasTime).stream().mapToDouble(IndicatorData::getValueF).sum();
+            values.add(String.format(CommonConstans.Twodecimalplaces, vl));
+        });
+
+        String max = String.format(CommonConstans.Twodecimalplaces, values.stream().mapToDouble(Double::parseDouble).max().getAsDouble());
+        String min = String.format(CommonConstans.Twodecimalplaces, values.stream().mapToDouble(Double::parseDouble).min().getAsDouble());
+        String mean = String.format(CommonConstans.Twodecimalplaces, values.stream().mapToDouble(Double::parseDouble).average().getAsDouble());
+        activePowerInfo.put("mean", mean);
+        activePowerInfo.put("max", max);
+        activePowerInfo.put("min", min);
+        activePowerInfo.put("maxTime", time.get(values.lastIndexOf(max)));
+        activePowerInfo.put("minTime", time.get(values.lastIndexOf(min)));
+        activePowerInfo.put("load", String.format(CommonConstans.Twodecimalplaces, Double.valueOf(values.get(values.size() - 1)) / installedCapacity.get()));
+
+
+        List<Map<String, Object>> seriesData = new ArrayList<>();
+        Map<String, Object> map3 = new HashMap<>();
+        Map<String, Object> map1 = new HashMap<>();
+        Map<String, Object> map2 = new HashMap<>();
+        map1.put("data", values);
+        seriesData.add(map1);
+        map.put("seriesData", seriesData);
+        map.put("axisData", time);
+        System.out.println(JSON.toJSONString(map));
+        try {
+            emqKeeper.getMqttClient().publish( "all_Power_table", JSON.toJSON(map).toString().getBytes("UTF-8"), 1, true);
+            emqKeeper.getMqttClient().publish("all_Power_info", JSON.toJSON(activePowerInfo).toString().getBytes("UTF-8"), 1, true);
+        } catch (Exception exception) {
+            exception.printStackTrace();
+        }
+        return map;
+    }
+
+    public  Map<String, Object> getDetailsWindqy(String s ) {
+        List<StationCacheInfoDto> stationCacheInfoDtos = getListStationCacheInfoDto();
+        AtomicReference<Double> installedCapacity = new AtomicReference<>(0.0);
+        stationCacheInfoDtos = stationCacheInfoDtos.stream().filter(stationCacheInfoDto -> stationCacheInfoDto.getBelongArea().equals(s)).collect(Collectors.toList());
+        stationCacheInfoDtos.forEach(stationCacheInfoDto -> {
+            installedCapacity.updateAndGet(v -> v + Double.parseDouble(stationCacheInfoDto.getInstalledCapacity()));
+        });
+        Map<String, Object> map = new HashMap<>();
+        List<String> values = new ArrayList<>();
+        List<String> fanGateWayIds = stationCacheInfoDtos.stream().filter(stationCacheInfoDto -> stationCacheInfoDto.getFanGatewayId()!=null).map(StationCacheInfoDto::getFanGatewayId).collect(Collectors.toList());
+        List<String> boostGateWayIds = stationCacheInfoDtos.stream().filter(stationCacheInfoDto -> stationCacheInfoDto.getBoosterGatewayId()!=null).map(StationCacheInfoDto::getBoosterGatewayId).collect(Collectors.toList());
+        fanGateWayIds.addAll(boostGateWayIds);
+       String gatewayIds =Optional.ofNullable(fanGateWayIds).orElse(new ArrayList<>()).stream().distinct().collect(Collectors.joining(","));
+        List<String> time = new ArrayList<>();
+        Map<String, Object> activePowerInfo = new HashMap<>();
+        List<IndicatorData> activePowerList = new ArrayList<>();
+        activePowerList = indicatorDataNewMapper.selectDataByequipmentIndexNameAndtimeqgNew(CommonConstans.ACTIVE_POWER,gatewayIds);
+        Map<String,List<IndicatorData>> stringObjectMap =activePowerList.stream().collect(Collectors.groupingBy(IndicatorData::getDasTime));
+        stringObjectMap.keySet().stream().forEach(dasTime -> {
+            time.add(dasTime.split(":")[1]);
+            Double vl=  stringObjectMap.get(dasTime).stream().mapToDouble(IndicatorData::getValueF).sum();
+            values.add(String.format(CommonConstans.Twodecimalplaces, vl));
+        });
+
+
+        String max = String.format(CommonConstans.Twodecimalplaces, values.stream().mapToDouble(Double::parseDouble).max().getAsDouble());
+        String min = String.format(CommonConstans.Twodecimalplaces, values.stream().mapToDouble(Double::parseDouble).min().getAsDouble());
+        String mean = String.format(CommonConstans.Twodecimalplaces, values.stream().mapToDouble(Double::parseDouble).average().getAsDouble());
+        activePowerInfo.put("mean", mean);
+        activePowerInfo.put("max", max);
+        activePowerInfo.put("min", min);
+        activePowerInfo.put("maxTime", time.get(values.lastIndexOf(max)));
+        activePowerInfo.put("minTime", time.get(values.lastIndexOf(min)));
+        activePowerInfo.put("load", String.format(CommonConstans.Twodecimalplaces, Double.valueOf(values.get(values.size() - 1)) / installedCapacity.get()));
+
+
+        List<Map<String, Object>> seriesData = new ArrayList<>();
+        Map<String, Object> map3 = new HashMap<>();
+        Map<String, Object> map1 = new HashMap<>();
+        Map<String, Object> map2 = new HashMap<>();
+        map1.put("data", values);
+        seriesData.add(map1);
+        map.put("seriesData", seriesData);
+        map.put("axisData", time);
+        try {
+            emqKeeper.getMqttClient().publish( s+"_Power_table", JSON.toJSON(map).toString().getBytes("UTF-8"), 1, true);
+            emqKeeper.getMqttClient().publish(s+"_Power_info", JSON.toJSON(activePowerInfo).toString().getBytes("UTF-8"), 1, true);
+        } catch (Exception exception) {
+            exception.printStackTrace();
+        }
+        return map;
+    }
 }

amos-boot-system-jxiop/amos-boot-module-jxiop-bigscreen-biz/src/main/java/com/yeejoin/amos/boot/module/jxiop/biz/tdanalysismapper/IndicatorDataNewMapper.java

+package com.yeejoin.amos.boot.module.jxiop.biz.tdanalysismapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.yeejoin.amos.boot.module.jxiop.biz.entity.IndicatorData;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Select;
+import org.springframework.stereotype.Repository;
+
+import java.util.List;
+
+@Repository
+public interface IndicatorDataNewMapper extends BaseMapper<IndicatorData> {
+
+
+    @Select("select LAST(address)address,LAST(gateway_id)gatewayId,LAST(data_type)dataType,LAST(equipment_index_name)equipmentIndexName,LAST(`value_f`)valueF,LAST(equipment_number)equipmentNumber,LAST(created_time)createdTime from iot_data.indicator_data where equipment_index_name =#{equipmentIndexName}  and    gateway_id =#{gatewayId} and  ts >=#{startTime} and ts <=#{endTime}    group by address ")
+    List<IndicatorData> selectlastfd(@Param("equipmentIndexName") String equipmentIndexName,@Param("gatewayId") String gatewayId,@Param("startTime") String startTime, @Param("endTime") String endTime);
+
+    @Select("select LAST(address)address,LAST(gateway_id)gatewayId,LAST(data_type)dataType,LAST(equipment_index_name)equipmentIndexName,LAST(`value_f`)valueF,LAST(equipment_number)equipmentNumber,LAST(created_time)createdTime from iot_data.indicator_data where   equipment_index_name =#{equipmentIndexName}  and    gateway_id =#{gatewayId} and  ts >=#{startTime} and ts <=#{endTime}    group by address ")
+    List<IndicatorData> selectlastgf(@Param("equipmentIndexName") String equipmentIndexName,@Param("gatewayId") String gatewayId,@Param("startTime") String startTime, @Param("endTime") String endTime);
+
+
+    @Select("select  created_time createdTime, `value_f` as valueF from iot_data.indicator_data where equipment_index_name =#{equipmentIndexName}  and  ts >= #{startTime} and ts <= #{endTime} and gateway_id =#{gatewayId}")
+    List<IndicatorData> selectDataByequipmentIndexNameAndtime(@Param("equipmentIndexName") String equipmentIndexName, @Param("startTime") String startTime, @Param("endTime") String endTime, @Param("gatewayId") String gatewayId);
+
+    @Select("select sum(valueFs) valueF,xtime  from (select  SUBSTR(`created_time`, 11,6) xtime, `value_f` as valueFs  from iot_data.indicator_data  where equipment_index_name =#{equipmentIndexName}  and  ts >= #{startTime} and ts <= #{endTime} and gateway_id in (#{gatewayId})) group  by  xtime  order by xtime   ")
+    List<IndicatorData> selectDataByequipmentIndexNameAndtimeqg(@Param("equipmentIndexName") String equipmentIndexName, @Param("startTime") String startTime, @Param("endTime") String endTime, @Param("gatewayId") String gatewayId);
+
+    @Select("select `value`, created_time, `value_f` as valueF from iot_data.indicator_data where point_name =#{pointName} and  ts >= TODAY()-8h and point_type ='SYNTHETIC'")
+    List<IndicatorData> selectDataByequipmentIndexNameAndtimeqgNew(@Param("equipmentIndexName") String pointName);
+
+    @Select("select `value`, created_time, `value_f` as valueF from iot_data.indicator_data where point_name =#{pointName} and  ts >= TODAY()-8h and gateway_id in (#{gatewayId})) and point_type ='SYNTHETIC'")
+    List<IndicatorData> selectDataByequipmentIndexNameAndtimeqgNew(@Param("equipmentIndexName") String pointName,@Param("gatewayId") String gatewayId);
+
+
+    @Select("select `value`, created_time, `value_f` as valueF from iot_data.indicator_data where equipment_index_name =#{equipmentIndexName} and equipment_number = #{equipmentNumber} and  ts >= #{startTime} and ts <= #{endTime} and gateway_id = #{gatewayId}")
+    List<IndicatorData> selectDataByequipmentIndexNameAndtimeAndEquipmentNumber(@Param("equipmentIndexName") String equipmentIndexName, @Param("equipmentNumber") String equipmentNumber, @Param("startTime") String startTime, @Param("endTime") String endTime, @Param("gatewayId") String gatewayId);
+
+    @Select("select `value`, created_time, `value_f` as valueF, equipment_index_name from iot_data.indicator_data where equipment_index_name like '%路电流%' and equipment_number = #{equipmentNumber} and  ts >= #{startTime} and ts <= #{endTime} and gateway_id =#{gatewayId}")
+    List<IndicatorData> selectDataByequipmentIndexNameAndtimeAndEquipmentNumberPv(@Param("equipmentNumber") String equipmentNumber, @Param("startTime") String startTime, @Param("endTime") String endTime, @Param("gatewayId") String gatewayId);
+
+
+
+}

amos-boot-system-jxiop/amos-boot-module-jxiop-bigscreen-biz/src/main/resources/application-dev.properties

@@ -11,13 +11,16 @@ spring.db2.datasource.url=jdbc:mysql://139.9.173.44:3306/jxiop_sync_data?allowMu
 spring.db2.datasource.username=root
 spring.db2.datasource.password=Yeejoin@2020
 spring.db2.datasource.driver-class-name: com.mysql.cj.jdbc.Driver
-
-
+##db3
 spring.db3.datasource.url=jdbc:TAOS-RS://139.9.170.47:6041/iot_data?user=root&password=taosdata&timezone=GMT%2b8&allowMultiQueries=true
 spring.db3.datasource.username=root
 spring.db3.datasource.password=taosdata
 spring.db3.datasource.driver-class-name: com.taosdata.jdbc.rs.RestfulDriver
-
+##db4
+spring.db4.datasource.url=jdbc:TAOS-RS://139.9.170.47:6041/iot_data?user=root&password=taosdata&timezone=GMT%2b8&allowMultiQueries=true
+spring.db4.datasource.username=root
+spring.db4.datasource.password=taosdata
+spring.db4.datasource.driver-class-name: com.taosdata.jdbc.rs.RestfulDriver
 
 ## eureka properties:
 eureka.instance.hostname=172.16.10.220

amos-boot-system-jxiop/amos-boot-module-jxiop-das/pom.xml

@@ -15,22 +15,6 @@
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     </properties>
     <dependencies>
-        <!--
-     <dependency>
-              <groupId>com.amosframework.boot</groupId>
-              <artifactId>amos-boot-module-ugp-api</artifactId>
-              <version>${amos-biz-boot.version}</version>
-          </dependency>-->
-<!--        <dependency>-->
-<!--            <groupId>com.amosframework.boot</groupId>-->
-<!--            <artifactId>amos-boot-module-common-biz</artifactId>-->
-<!--            <version>${amos-biz-boot.version}</version>-->
-<!--        </dependency>-->
-<!--        <dependency>-->
-<!--            <groupId>com.amosframework.boot</groupId>-->
-<!--            <artifactId>amos-boot-biz-common</artifactId>-->
-<!--            <version>1.0.0</version>-->
-<!--        </dependency>-->
         <dependency>
             <groupId>com.alibaba</groupId>
             <artifactId>druid-spring-boot-starter</artifactId>
@@ -39,6 +23,7 @@
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-jdbc</artifactId>
+            <version>2.1.6.RELEASE</version>
         </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
@@ -66,6 +51,11 @@
             <artifactId>fastjson</artifactId>
             <version>1.2.47</version>
         </dependency>
+        <dependency>
+            <groupId>cn.hutool</groupId>
+            <artifactId>hutool-all</artifactId>
+            <version>5.8.16</version>
+        </dependency>
     </dependencies>
 
     <build>

amos-boot-system-jxiop/amos-boot-module-jxiop-das/src/main/java/com/yeejoin/amos/AmosJxiopDasApplication.java

@@ -31,7 +31,7 @@ import java.net.InetAddress;
 @EnableAsync
 @EnableScheduling
 @SpringBootApplication(exclude = {DataSourceAutoConfiguration.class, DruidDataSourceAutoConfigure.class})
-@MapperScan({"com.yeejoin.amos.boot.module.das.mapper.msyql","com.yeejoin.amos.boot.module.das.mapper.tdengineanalysis","com.yeejoin.amos.boot.module.das.mapper.tdengineiot"})
+@MapperScan({"com.yeejoin.amos.boot.module.das.mapper.msyql","com.yeejoin.amos.boot.module.das.mapper.analysis","com.yeejoin.amos.boot.module.das.mapper.iot"})
 @ComponentScan({"springfox.documentation.schema", "com.yeejoin.amos.boot.module.das","com.yeejoin.amos.boot.module.das.service.impl","org.typroject.tyboot.component"})
 public class AmosJxiopDasApplication {
 

amos-boot-system-jxiop/amos-boot-module-jxiop-das/src/main/java/com/yeejoin/amos/boot/module/das/entity/analysis/IndicatorData.java

@@ -16,9 +16,11 @@ public class IndicatorData {
     private Date createdTime;
     private String gatewayId;
     private String dataType;
+    private String dasTime;
     private String pointSeq;
     private String pointAddress;
     private String pointLocation;
+    private String pointType;
     private String pointName;
     private String value="0";
     private Double valueF;

amos-boot-system-jxiop/amos-boot-module-jxiop-das/src/main/java/com/yeejoin/amos/boot/module/das/mapper/analysis/IndicatorDataMapper.java

@@ -8,6 +8,6 @@ import java.util.List;
 
 @Component
 public interface IndicatorDataMapper {
-    int insertBatch(@Param("list") List<IndicatorData> list, @Param("gatewayId")String gatewayId);
+    int insertBatch(@Param("list") List<IndicatorData> list, @Param("gatewayId")String gatewayId,@Param("dasTime")String dasTime);
     void createTable();
 }

amos-boot-system-jxiop/amos-boot-module-jxiop-das/src/main/java/com/yeejoin/amos/boot/module/das/mapper/mysql/FrontGatewayDevicePointsMapper.java

@@ -8,6 +8,6 @@ import java.util.List;
 public interface FrontGatewayDevicePointsMapper extends BaseMapper<FrontGatewayDevicePoints> {
     @Select("select distinct gateway_id from iot_front_gateway_device_points")
     List<String> getGatewayIds();
-    @Select("select SEQUENCE_NBR,POINT_NAME,POINT_DATA_TYPE,POINT_ADDRESS,POINT_LOCATION from iot_front_gateway_device_points where gateway_id = #{gatewayId}")
+    @Select("select SEQUENCE_NBR,POINT_NAME,POINT_DATA_TYPE,POINT_ADDRESS,POINT_LOCATION,POINT_TYPE,DATA_TYPE from iot_front_gateway_device_points where gateway_id = #{gatewayId}")
     List<FrontGatewayDevicePoints> getFrontGatewayDevicePointsByGatewayId(String gatewayId);
 }

amos-boot-system-jxiop/amos-boot-module-jxiop-das/src/main/java/com/yeejoin/amos/boot/module/das/service/impl/DasServiceImpl.java

 package com.yeejoin.amos.boot.module.das.service.impl;
 
+import cn.hutool.core.date.DateUtil;
 import com.alibaba.fastjson.JSON;
 import com.google.common.collect.Lists;
 
@@ -18,6 +19,7 @@ import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 import org.springframework.util.ObjectUtils;
 import org.typroject.tyboot.component.emq.EmqKeeper;
+
 import java.util.*;
 import java.util.stream.Collectors;
 
@@ -41,16 +43,17 @@ public class DasServiceImpl implements DasService {
      * 完成后，会记录此次操作所花费的时间。
      */
     public void dataSolidification() {
-
+        String dasTime = DateUtil.format(new Date(), "yyyy-MM-dd HH:mm:00");
+        log.info("数据采集开始执行-采集时间：：" + dasTime );
         // 记录操作开始时间
-        Long startTime = System.currentTimeMillis();
+        Long  startTime = System.currentTimeMillis();
         // 创建新表
         indicatorDataMapper.createTable();
         // 获取所有网关ID
         List<String> gateWayIds = frontGatewayDevicePointsMapper.getGatewayIds();
         // 并行处理每个网关ID的数据凝固
         gateWayIds.parallelStream().forEach(gatewayId -> {
-            dataSolidificationByGatewayId(gatewayId);
+            dataSolidificationByGatewayId(gatewayId,dasTime);
         });
         // 记录操作结束时间
         Long endTime = System.currentTimeMillis();
@@ -66,7 +69,7 @@ public class DasServiceImpl implements DasService {
      * @param gatewayId 网关的唯一标识符，用于查询相关设备点信息和数据点值。
      */
     @Async("jxiopAsyncExecutor")
-    public void dataSolidificationByGatewayId(String gatewayId) {
+    public void dataSolidificationByGatewayId(String gatewayId,String dasTime) {
         // 根据网关ID查询设备点信息
         List<FrontGatewayDevicePoints> tempPoints = frontGatewayDevicePointsMapper.getFrontGatewayDevicePointsByGatewayId(gatewayId);
         if (!ObjectUtils.isEmpty(tempPoints)) {
@@ -90,6 +93,7 @@ public class DasServiceImpl implements DasService {
                     indicatorData.setPointAddress(point.getPointAddress());
                     indicatorData.setPointLocation(point.getPointLocation());
                     indicatorData.setPointName(point.getPointName());
+                    indicatorData.setPointType(point.getPointType());
                     // 设置数据点的值，如果是布尔值则进行转换
                     indicatorData.setValue(stbMap.get(point.getSequenceNbr().toString()));
                     if (!ObjectUtils.isEmpty(indicatorData.getValue()) && !booleans.contains(indicatorData.getValue())) {
@@ -104,10 +108,9 @@ public class DasServiceImpl implements DasService {
                 // 批量插入构建的数据模型到数据库
                 Lists.partition(listAll, 1000).stream().forEach(
                         list -> {
-                            indicatorDataMapper.insertBatch(list, gatewayId);
+                            indicatorDataMapper.insertBatch(list, gatewayId,dasTime);
                         }
                 );
-
                 // 向EMQX发送消息，通知数据同步成功
                 try {
                     HashMap<String, String> syncFlag = new HashMap<>();

amos-boot-system-jxiop/amos-boot-module-jxiop-das/src/main/resources/application-dev.properties

@@ -31,6 +31,8 @@ spring.redis.password=yeejoin@2020
 emqx.clean-session=true
 emqx.client-id=${spring.application.name}-${random.int[1024,65536]}
 emqx.broker=tcp://10.20.1.210:2883
+emqx.client-user-name=admin
+emqx.client-password=public
 emqx.user-name=admin
 emqx.password=public
 mqtt.scene.host=mqtt://10.20.1.210:8083/mqtt

amos-boot-system-jxiop/amos-boot-module-jxiop-das/src/main/resources/mapper/analysis/IndicatorDataMapper.xml

@@ -8,11 +8,13 @@
     <update id="createTable">
         create STABLE if not exists indicator_data
                (created_time timestamp,
+               das_time  VARCHAR(100),
                data_type NCHAR(12),
                point_seq VARCHAR(100) ,
                point_address VARCHAR(100) ,
                point_location VARCHAR(500) ,
                point_name VARCHAR(200),
+               point_type VARCHAR(50),
                `value` VARCHAR(50),
                `value_f` float)
         TAGS  (gateway_id binary(64));
@@ -24,11 +26,13 @@
             indicator_data_#{gatewayId,jdbcType=VARCHAR} USING indicator_data
             TAGS (#{gatewayId,jdbcType=VARCHAR})
             VALUES (NOW + #{index}a,
+            #{dasTime,jdbcType=VARCHAR},
             #{item.dataType,jdbcType=VARCHAR},
             #{item.pointSeq,jdbcType=VARCHAR},
             #{item.pointAddress,jdbcType=VARCHAR},
             #{item.pointLocation,jdbcType=VARCHAR},
             #{item.pointName,jdbcType=VARCHAR},
+            #{item.pointType,jdbcType=VARCHAR},
             #{item.value,jdbcType=VARCHAR},
             #{item.valueF,jdbcType=FLOAT})
         </foreach>