社会贡献使用消息发送

amos-boot-system-jxiop/amos-boot-module-jxiop-monitor-biz/src/main/java/com/yeejoin/amos/boot/module/jxiop/biz/dto/IndicatorsDto.java

+package com.yeejoin.amos.boot.module.jxiop.biz.dto;
+
+import lombok.Data;
+
+@Data
+public class IndicatorsDto {
+    
+}

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

@@ -10,17 +10,25 @@ import com.yeejoin.amos.boot.module.jxiop.api.mapper.StationBasicMapper;
 import com.yeejoin.amos.boot.module.jxiop.api.mapper.StationCoordinateMapper;
 import com.yeejoin.amos.boot.module.jxiop.biz.dto.RegionNationWideDto;
 import com.yeejoin.amos.boot.module.jxiop.biz.dto.SocialContributionDto;
+import com.yeejoin.amos.boot.module.jxiop.biz.utils.InfluxDButils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.scheduling.annotation.Async;
+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.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicReference;
 
+@Async
 @Service
 public class MonitoringServiceImpl {
+    Logger logger = LoggerFactory.getLogger(MonitoringServiceImpl.class);
     /**
      * 区域mapper
      */
@@ -50,8 +58,14 @@ public class MonitoringServiceImpl {
 
     @Autowired
     MonitorFanIndicatorImpl monitorFanIndicatorImpl;
+    @Autowired
+    EmqKeeper emqKeeper;
+    @Autowired
+    InfluxDButils influxDButils;
+
     /**
      * 根据省份名称查询电站详情
+     *
      * @param provinceName
      * @return
      */
@@ -69,7 +83,7 @@ public class MonitoringServiceImpl {
                 Region region = regionList.get(i);
                 List<StationBasic> list = stationBasicMapper.selectList(new QueryWrapper<StationBasic>().like("belong_area",region.getRegionCode()));
                 list.forEach(stationBasic -> {
-                    if(!ObjectUtils.isEmpty(stationBasic.getFanGatewayId())&&!ObjectUtils.isEmpty(stationBasic.getStationNumber())){
+                    if (!ObjectUtils.isEmpty(stationBasic.getGatewayId()) && !ObjectUtils.isEmpty(stationBasic.getStationNumber())) {
                         regionNationWideDto.setJsJoin(true);
                     }
                     if(stationBasic.getStationType().equals("FDZ")){
@@ -97,8 +111,8 @@ public class MonitoringServiceImpl {
                 regionNationWideDto.setStaitionName(stationBasic.getStationName());
                 regionNationWideDto.setStaitionType(stationBasic.getStationType());
                 String speendOrirradiate = "";
-                if(stationBasic.getStationType().equals("FDZ")){
-                    speendOrirradiate = String.valueOf( getIndicatoralueAvage(stationBasic.getFanGatewayId(),"30秒平均风速"));
+                if (stationBasic.getStationType().equals("FDZ")) {
+                    speendOrirradiate = String.valueOf(getIndicatoralueAvage(stationBasic.getGatewayId(), "30秒平均风速"));
                     regionNationWideDto.setSpeendOrirradiate(speendOrirradiate);
                 }else{
                     regionNationWideDto.setSpeendOrirradiate(String.valueOf(Math.random()*1000));
@@ -116,50 +130,58 @@ public class MonitoringServiceImpl {
 
     /**
      * 根据场站编号获取该场站的装机容量
+     *
      * @param werks
      * @return
      */
-    public Double getStationCaPACITYL(String werks){
-       return this.SjglZsjZsbtzServiceImpl.getStationCapactityByStationWerks(werks);
+    public Double getStationCaPACITYL(String werks) {
+        return this.SjglZsjZsbtzServiceImpl.getStationCapactityByStationWerks(werks);
     }
 
     /**
      * 根据网关id以及指标值 获取同一网关下指标的平均值
-     * @param gateway 网关id
+     *
+     * @param gateway        网关id
      * @param indicatorcName 指标名称
      * @return 同一网关下指标的平均值
      */
-    public Object getIndicatoralueAvage (String gateway ,String indicatorcName){
-        return this.monitorFanIndicatorImpl.getIndicatoralueAvage(gateway,indicatorcName);
+    public Object getIndicatoralueAvage(String gateway, String indicatorcName) {
+        return this.monitorFanIndicatorImpl.getIndicatoralueAvage(gateway, indicatorcName);
     }
+
     /**
      * 根据网关id以及指标值 同一网关下指标的值总和
-     * @param gateway 网关id
+     *
+     * @param gateway        网关id
      * @param indicatorcName 指标名称
      * @return 同一网关下指标的值总和
      */
-    public Object getIndicatoralueTotal (String gateway ,String indicatorcName){
-        return this.monitorFanIndicatorImpl.getIndicatoralueTotal(gateway,indicatorcName);
+    public Object getIndicatoralueTotal(String gateway, String indicatorcName) {
+        return this.monitorFanIndicatorImpl.getIndicatoralueTotal(gateway, indicatorcName);
     }
 
-    public List<SocialContributionDto>  getTotalSocialContribution(){
-        List<SocialContributionDto> socialContributionDtoList =new ArrayList<>();
+    @Scheduled(cron = "*/10 * * * * *")
+    public void getTotalSocialContribution() {
+        Page<SocialContributionDto> socialContributionDtoPage = new Page<SocialContributionDto>();
+        logger.error("--------------------------社会贡献定时执行----------------------------------------------");
+        List<SocialContributionDto> socialContributionDtoList = new ArrayList<>();
+        //定义社会贡献列表
         AtomicReference<Double> totalSocialContribution = new AtomicReference<>(0.0);
         //获取所有网关id不为空的数据
         List<StationBasic> stationBasicList = stationBasicMapper.selectList(new QueryWrapper<StationBasic>().isNotNull("gateway_id"));
         stationBasicList.forEach(stationBasic -> {
-            totalSocialContribution.updateAndGet(v -> v + (Double) monitorFanIndicatorImpl.getIndicatoralueTotal(stationBasic.getFanGatewayId(), "年发电量"));
+            totalSocialContribution.updateAndGet(v -> v + (Double) monitorFanIndicatorImpl.getIndicatoralueTotal(stationBasic.getGatewayId(), "年发电量"));
         });
         SocialContributionDto co2 = new SocialContributionDto();
         SocialContributionDto coal = new SocialContributionDto();
         SocialContributionDto toner = new SocialContributionDto();
         SocialContributionDto so2 = new SocialContributionDto();
-       /** 社会贡献原始计算公式，后边的数据已经经过处理
-        //二氧化碳计算公式  *10000*832/10000/1000/1000
-        //标准煤计算公式  *10000*304.9/10000/1000/1000
-        //碳粉尘计算公式 *10000*0.032/1000/1000
-        //二氧化硫计算公式  =A2*10000*0.16/1000/1000
-        */
+        /** 社会贡献原始计算公式，后边的数据已经经过处理
+         //二氧化碳计算公式  *10000*832/10000/1000/1000
+         //标准煤计算公式  *10000*304.9/10000/1000/1000
+         //碳粉尘计算公式 *10000*0.032/1000/1000
+         //二氧化硫计算公式  =A2*10000*0.16/1000/1000
+         */
         co2.setUnit("二氧化碳减排量（万t）");
         co2.setTitle(totalSocialContribution.get() * 0.000832);
         socialContributionDtoList.add(co2);
@@ -172,6 +194,16 @@ public class MonitoringServiceImpl {
         so2.setUnit("二氧化硫减排量（万t）");
         so2.setTitle(totalSocialContribution.get() * 0.0016);
         socialContributionDtoList.add(so2);
-        return  socialContributionDtoList;
+        socialContributionDtoPage.setRecords(socialContributionDtoList);
+        socialContributionDtoPage.setTotal(100);
+        socialContributionDtoPage.setCurrent(1);
+        try {
+            emqKeeper.getMqttClient().publish("SocialContribution", JSON.toJSON(socialContributionDtoPage).toString().getBytes("UTF-8"), 1, true);
+            logger.info("-----------------发送社会贡献消息=================== 失败！"+JSON.toJSONString(socialContributionDtoPage));
+        } catch (Exception exception) {
+            logger.error("-----------------发送社会贡献消息=================== 失败！");
+        }
     }
+
+
 }