focus/ipfcst-console/src/main/java/com/jiayue/ipfcst/console/service/ReceiveDataService.java

package com.jiayue.ipfcst.console.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.jiayue.ipfcst.common.core.util.DateTimeUtil;
import com.jiayue.ipfcst.common.data.constant.enums.EquipmentTypeEnum;
import com.jiayue.ipfcst.common.data.entity.*;
import com.jiayue.ipfcst.common.data.repository.*;
import com.jiayue.ipfcst.console.util.RedisUtils;
import org.apache.http.HttpEntity;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.util.EntityUtils;
import org.springframework.beans.factory.annotation.Value;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;


import java.io.IOException;
import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * 接收数据service类
 *
 * @author whc
 */
@Service
@Slf4j
public class ReceiveDataService {
  @Autowired
  ElectricFieldService electricFieldService;
  @Autowired
  WindTowerInfoService windTowerInfoService;
  @Autowired
  WindTurbineInfoService windTurbineInfoService;
  @Autowired
  WeatherStationInfoService weatherStationInfoService;
  @Autowired
  InverterInfoService inverterInfoService;
  @Autowired
  DataPointService dataPointService;
  @Autowired
  WeatherStationStatusDataRepository weatherStationStatusDataRepository;
  @Autowired
  InverterStatusDataRepository inverterStatusDataRepository;
  @Autowired
  WindTurbineStatusDataRepository windTurbineStatusDataRepository;
  @Autowired
  WindTowerStatusDataRepository windTowerStatusDataRepository;
  @Autowired
  RedisUtils redisUtils;
  @Autowired
  EquipmentAttributeService equipmentAttributeService;
  @Autowired
  PowerStationStatusDataRepository powerStationStatusDataRepository;
  @Autowired
  PowerStationDataPackerContainer powerStationDataPackerContainer;

  @Value("${receive.ip}")
  private String ip;
  @Value("${receive.port}")
  private  String port;
  @Value("${receive.path}")
  private  String path;
  private  final String Active = "activePower";

  //请求超时时间,这个时间定义了socket读数据的超时时间，也就是连接到服务器之后到从服务器获取响应数据需要等待的时间,发生超时，会抛出SocketTimeoutException异常。
  private static final int SOCKET_TIME_OUT = 5000;
  //连接超时时间,这个时间定义了通过网络与服务器建立连接的超时时间，也就是取得了连接池中的某个连接之后到接通目标url的连接等待时间。发生超时，会抛出ConnectionTimeoutException异常
  private static final int CONNECT_TIME_OUT = 3000;


  public void receive() {
    List<ElectricField> electricFieldList = electricFieldService.getAll();
    //每个场站请求一次
    for (ElectricField electricField : electricFieldList) {
      log.info(electricField.getName() + "开始请求数据");
      Long startTime = DateTimeUtil.getCurrentTimeForMinute().getTime();
      Long endTime = startTime + 60000L;
      try {
        //通用查询
        if (electricField.getElectricFieldTypeEnum().getCode() == 1) {
          //气象站
          List<WeatherStationInfo> weatherStationInfoList = weatherStationInfoService.get(electricField.getStationCode());
          //逆变器
          List<InverterInfo> inverterInfoList = inverterInfoService.getByStationCode(electricField.getStationCode());
          //按设备请求
          for (WeatherStationInfo weatherStationInfo : weatherStationInfoList) {
            map(electricField, startTime, endTime, weatherStationInfo.getEquipmentNo(), weatherStationInfo.getId(), weatherStationInfo.getEquipmentType());
          }
          for (InverterInfo inverterInfo : inverterInfoList) {
            map(electricField, startTime, endTime, inverterInfo.getEquipmentNo(), inverterInfo.getId(), inverterInfo.getEquipmentType());
          }
        } else {
          //测风塔
          List<WindTowerInfo> windTowerInfoList = windTowerInfoService.get(electricField.getStationCode());
          //风机
          List<WindTurbineInfo> windTurbineInfoList = windTurbineInfoService.getByStationCode(electricField.getStationCode());
          for (WindTowerInfo windTowerInfo : windTowerInfoList) {
            map(electricField, startTime, endTime, windTowerInfo.getEquipmentNo(), windTowerInfo.getId(), windTowerInfo.getEquipmentType());
          }
          for (WindTurbineInfo windTurbineInfo : windTurbineInfoList) {
            map(electricField, startTime, endTime, windTurbineInfo.getEquipmentNo(), windTurbineInfo.getId(), windTurbineInfo.getEquipmentType());
          }
        }
        savePowerStationStatusData(electricField);
      } catch (Exception e) {
        e.printStackTrace();
      }
    }
  }

  private void map(ElectricField electricField, Long startTime, Long endTime, String equipmentNo, Integer id, EquipmentTypeEnum equipmentType) {
    HashMap<String, Object> paramMap = new HashMap<>();
    paramMap.put("deviceIds", equipmentNo);
    paramMap.put("startTime", startTime);
    paramMap.put("endTime", endTime);
    log.info("param"+JSON.toJSONString(paramMap));
    String body = httpClient(paramMap);
    AnalysisData(body, id, equipmentType, electricField);
  }

  /**
   * 解析数据
   *
   * @param body          所有数据
   * @param equipmentId   设备id
   * @param equipmentType 设备类型
   * @param electricField 场站对象
   */
  public void AnalysisData(String body, Integer equipmentId, EquipmentTypeEnum equipmentType, ElectricField electricField) {
    try {
      List<DataPoint> dataPointList = dataPointService.getByEquipmentType(equipmentType);
      JSONObject jsonObject= JSON.parseObject(body);
      JSONArray jsonResults = (JSONArray) jsonObject.get("results");
      if (jsonResults.size()>0){
        JSONObject jsonObj = jsonResults.getJSONObject(0);
        JSONArray jsonRows = (JSONArray) jsonObj.get("rows");
        //所有设备点位
        JSONArray sensorIds = (JSONArray) jsonObj.get("sensorIds");
        JSONObject values;
        Map<String, String> map = new HashMap<>();
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        if (jsonRows.size()>0){
          values = jsonRows.getJSONObject(0);
          //时间戳
          String time = values.getString("timestamp");
          //所有数据
          JSONArray jsonValues = (JSONArray) values.get("values");
          long timestamp = Long.parseLong(time);
          map.put("stationCode",electricField.getStationCode());
          map.put("equipmentId", equipmentId.toString());
          map.put("time", simpleDateFormat.format(timestamp));
          for (DataPoint dataPoint : dataPointList) {
            for (int i = 0; i < sensorIds.size(); i++) {
              if (dataPoint.getMeasuringPoint().equals(sensorIds.get(i))) {
                if(jsonValues.get(i) == null ){
                  map.put(dataPoint.getEquipmentAttribute().getFieldName(), "-99");
                }else{
                  map.put(dataPoint.getEquipmentAttribute().getFieldName(), jsonValues.get(i).toString());
                }
              }
            }
          }
          String STATUS = "status";
          switch (map.get(STATUS)) {
            case "1.00":
            case "true":
              //运行
              map.put(STATUS, "1");
              break;
            case "2.00":
              //待机
              map.put(STATUS, "2");
              break;
            case "3.00":
              //停用
              map.put(STATUS, "3");
              break;
            case "4.00":
            case "false":
              //故障
              map.put(STATUS, "4");
              break;
            case "-99":
              map.put(STATUS, "5");
              break;
            default:
              break;
          }

          String ap = map.get(Active);
          switch(equipmentType.getCode()){
            case 1:
              //气象站
              redisUtils.hmset("qxz-" + electricField.getStationCode() + "-" + equipmentId, map);
              WeatherStationStatusData weatherStationStatusData = JSON.parseObject(JSON.toJSONString(map), WeatherStationStatusData.class);
              weatherStationStatusDataRepository.save(weatherStationStatusData);
              log.info("qxz-" + electricField.getStationCode() + "-" + equipmentId+"已存入");
              break;
            case 2:
              //逆变器
              if (ap != null && map.get("electricalCurrent") != null) {
                //有功
                BigDecimal activePower = new BigDecimal(ap);
                //电流
                BigDecimal electricalCurrent = new BigDecimal(map.get("electricalCurrent"));
                //有功/电流=电压
                BigDecimal voltage = activePower.divide(electricalCurrent, 2, BigDecimal.ROUND_HALF_UP);
                map.put("voltage", voltage.toString());
              }
              redisUtils.hmset("nbq-" + electricField.getStationCode() + "-" + equipmentId, map);
              InverterStatusData inverterStatusData = JSON.parseObject(JSON.toJSONString(map), InverterStatusData.class);
              inverterStatusDataRepository.save(inverterStatusData);
              log.info("nbq-" + electricField.getStationCode() + "-" + equipmentId+"已存入");
              break;
            case 3:
              redisUtils.hmset("fj-" + electricField.getStationCode() + "-" + equipmentId, map);
              WindTurbineStatusData windTurbineStatusData = JSON.parseObject(JSON.toJSONString(map), WindTurbineStatusData.class);
              windTurbineStatusDataRepository.save(windTurbineStatusData);
              log.info("fj-" + electricField.getStationCode() + "-" + equipmentId+"已存入");
              break;
            case 4:
              redisUtils.hmset("cft-" + electricField.getStationCode() + "-" + equipmentId, map);
              WindTowerStatusData windTowerStatusData = JSON.parseObject(JSON.toJSONString(map), WindTowerStatusData.class);
              windTowerStatusDataRepository.save(windTowerStatusData);
              log.info("cft-" + electricField.getStationCode() + "-" + equipmentId+"已存入");
              break;
            default:
              break;
          }
        }else {
          log.info(electricField.getStationCode()+ "-" +equipmentType.getMessage() + "-" + equipmentId+"无数据，不进行任何操作");
        }
      }else {
        log.info(electricField.getStationCode()+ "-" +equipmentType.getMessage() + "-" + equipmentId+"对方没有该设备，请核对设备编号");
      }
    } catch (Exception e) {
      log.info(electricField.getName()+equipmentType.getMessage()+"接数程序异常");
      e.printStackTrace();
    }
  }

  /**
   * 发http请求
   * @param paramMap 参数
   * @return 返回的数据
   */
  public String httpClient( HashMap<String, Object> paramMap){
    String body = "";
    try{
      CloseableHttpClient  httpClient = HttpClientBuilder.create().build();
      HttpPost httpPost = new HttpPost("http://"+ip+":"+port+path);

      StringEntity entity = new StringEntity(JSON.toJSONString(paramMap),"UTF-8");
      httpPost.setEntity(entity);
      //设置请求超时时间，链接超时时间
      RequestConfig reqConfig = RequestConfig.custom().setSocketTimeout(SOCKET_TIME_OUT).setConnectTimeout(CONNECT_TIME_OUT).build();
      httpPost.setConfig(reqConfig);
      httpPost.setHeader("Content-Type", "application/json; charset=utf-8");
      CloseableHttpResponse response;
      response = httpClient.execute(httpPost);
      HttpEntity responseEntity = response.getEntity();
      body = EntityUtils.toString(responseEntity, StandardCharsets.UTF_8);
    }catch (RuntimeException | IOException e){
      e.printStackTrace();
      log.info("请求异常");
      log.info("所用线程"+Thread.currentThread().getName());
      log.info("停止请求数据");
      Thread.currentThread().stop();
    }
    return body;
  }
  public void savePowerStationStatusData(ElectricField electricField) {
    try {
      log.info("开始计算实际功率");
      BigDecimal pssd = new BigDecimal("0");
      if (electricField.getElectricFieldTypeEnum().getCode() == 1) {
        List<InverterInfo> inverterInfoList = inverterInfoService.getByStationCode(electricField.getStationCode());
        for (InverterInfo inverterInfo : inverterInfoList) {
          Map<String, String> getMap = redisUtils.hgetall("nbq-" + electricField.getStationCode() + "-" + inverterInfo.getId());
          String activePower = getMap.get(Active);
          if (activePower != null) {
            pssd = pssd.add(new BigDecimal(activePower));
          }
        }
      } else {
        List<WindTurbineInfo> windTurbineInfoList = windTurbineInfoService.getByStationCode(electricField.getStationCode());
        for (WindTurbineInfo windTurbineInfo : windTurbineInfoList) {
          Map<String, String> getMap = redisUtils.hgetall("fj-" + electricField.getStationCode() + "-" + windTurbineInfo.getId());
          String activePower = getMap.get(Active);
          if (activePower != null) {
            pssd = pssd.add(new BigDecimal(activePower));
          }
        }
      }
      if (pssd.compareTo(new BigDecimal("0")) > 0) {
        pssd = pssd.divide(new BigDecimal("1000"), 2, BigDecimal.ROUND_HALF_UP);
      }
      PowerStationStatusData p = powerStationDataPackerContainer.getDataPacker(electricField.getStationCode()).packageData(pssd);
      powerStationStatusDataRepository.save(p);
      //对象转map
      Map<String, String> map    = JSON.parseObject(JSON.toJSONString(p), new TypeReference<Map<String, String>>() {
      });
      redisUtils.hmset("power-" + electricField.getStationCode(), map);
      log.info("实际功率结束");
    } catch (Exception e) {
      e.printStackTrace();
    }
  }
  public  void stop(){
    Thread thread = new Thread(() -> {
      while (Thread.currentThread().isInterrupted()) {
        log.info("停止请求数据");
      }
    });
    thread.start();
    try {
      Thread.sleep(1);
    } catch (Exception e) {
      e.printStackTrace();
    }
    thread.interrupt();
  }
}