社会贡献改用消息发送且计算从时序库中获取

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

@@ -2,7 +2,25 @@ package com.yeejoin.amos.boot.module.jxiop.biz.dto;
 
 import lombok.Data;
 
+/**
+ *  指标dto-用于接收influxdb里边查询的到的数据
+ */
 @Data
 public class IndicatorsDto {
-    
+    private String address;
+    private String createdTime;
+    private String dataType;
+    private String equipmentIndex;
+    private String equipmentIndexName;
+    private String equipmentNumber;
+    private String equipmentSpecificName;
+    private String equipmentsIdx;
+    private String frontModule;
+    private String gatewayId;
+    private String isAlarm;
+    private String systemType;
+    private String traceId;
+    private String unit;
+    private String value;
+    private String valueLabel;
 }

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

+package com.yeejoin.amos.boot.module.jxiop.biz.service.impl;
+
+import com.yeejoin.amos.boot.module.jxiop.biz.dto.IndicatorsDto;
+import com.yeejoin.amos.boot.module.jxiop.biz.utils.InfluxDButils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+@Service
+public class CommonServiceImpl {
+    @Autowired
+    InfluxDButils influxDButils;
+
+    /**
+     * @deprecated  获取指标值总和
+     * @param gatewayId 网关id 用于拼接sql语句
+     * @param indicator 指标名称  查询条件-根据指标名称获取风速
+     * @return 指标值总和
+     */
+    public Double getTotalByIndicatior(String gatewayId,String indicator){
+        String sql = "SELECT * FROM indicators_"+gatewayId+" where equipmentIndexName='"+indicator+"'";
+        Double totalvalue =  0.0;
+        List<IndicatorsDto> indicatorsDtoList = influxDButils.getListData(sql,IndicatorsDto.class);
+        List<Double> doubleList  = indicatorsDtoList.stream().map(indicatorsDto -> Double.parseDouble(indicatorsDto.getValue())).collect(Collectors.toList());
+        totalvalue = doubleList.stream().mapToDouble(Double::doubleValue).sum();
+        return totalvalue;
+    }
+    /**
+     * @deprecated  获取指标值平均值
+     * @param gatewayId 网关id 用于拼接sql语句
+     * @param indicator 指标名称  查询条件-根据指标名称获取风速
+     * @return 指标值总和
+     */
+    public Double getAvgvalueByIndicatior(String gatewayId,String indicator){
+        String sql = "SELECT * FROM indicators_"+gatewayId+" where equipmentIndexName='"+indicator+"'";
+        Double avageValue =  0.0;
+        List<IndicatorsDto> indicatorsDtoList = influxDButils.getListData(sql,IndicatorsDto.class);
+        List<Double> doubleList  = indicatorsDtoList.stream().map(indicatorsDto -> Double.parseDouble(indicatorsDto.getValue())).collect(Collectors.toList());
+        avageValue = doubleList.stream().mapToDouble(Double::doubleValue).average().getAsDouble();
+        return avageValue;
+    }
+}

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

+package com.yeejoin.amos.boot.module.jxiop.biz.service.impl;
+
+import com.alibaba.fastjson.JSON;
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import com.yeejoin.amos.boot.module.jxiop.api.entity.Region;
+import com.yeejoin.amos.boot.module.jxiop.api.entity.StationBasic;
+import com.yeejoin.amos.boot.module.jxiop.api.entity.StationCoordinate;
+import com.yeejoin.amos.boot.module.jxiop.api.mapper.RegionMapper;
+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.IndicatorsDto;
+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;
+import java.util.stream.Collectors;
+
+@Async
+@Service
+public class MonitoringServiceIMQTTmpl {
+    Logger logger = LoggerFactory.getLogger(MonitoringServiceIMQTTmpl.class);
+
+    /**
+     * 场站mapper
+     */
+    @Autowired
+    StationBasicMapper stationBasicMapper;
+
+    @Autowired
+    CommonServiceImpl commonServiceImpl;
+    @Autowired
+    EmqKeeper emqKeeper;
+
+
+    @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("fan_gateway_id"));
+        stationBasicList.forEach(stationBasic -> {
+            totalSocialContribution.updateAndGet(v -> v + commonServiceImpl.getTotalByIndicatior(stationBasic.getFanGatewayId(), "年发电量"));
+        });
+        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
+         */
+        co2.setUnit("二氧化碳减排量（万t）");
+        co2.setTitle(totalSocialContribution.get() * 0.000832);
+        socialContributionDtoList.add(co2);
+        coal.setUnit("节约标准煤（万t）");
+        coal.setTitle(totalSocialContribution.get() * 0.0003049);
+        socialContributionDtoList.add(coal);
+        toner.setUnit("碳粉尘减排量（万t）");
+        toner.setTitle(totalSocialContribution.get() * 0.00032);
+        socialContributionDtoList.add(toner);
+        so2.setUnit("二氧化硫减排量（万t）");
+        so2.setTitle(totalSocialContribution.get() * 0.0016);
+        socialContributionDtoList.add(so2);
+        socialContributionDtoPage.setRecords(socialContributionDtoList);
+        socialContributionDtoPage.setTotal(100);
+        socialContributionDtoPage.setCurrent(1);
+        try {
+            emqKeeper.getMqttClient().publish("socialcontribution_topic", JSON.toJSON(socialContributionDtoPage).toString().getBytes("UTF-8"), 1, true);
+            logger.info("-----------------发送社会贡献消息=================== 成功！" + JSON.toJSONString(socialContributionDtoPage));
+        } catch (Exception exception) {
+            logger.error("-----------------发送社会贡献消息=================== 失败！");
+        }
+    }
+
+}