SyncTool/jb_syncdata
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1.分布式光伏遥测数据接口新增

Browse Source
This commit is contained in:
wr
2024-06-20 09:44:42 +08:00
parent e1b216fbdc
commit 6c8d5d3ff7
16 changed files with 605 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
src/main/java/com/njcn/jbsyncdata/controller/DisPhotovoltaicController.java
View File

@@ -8,12 +8,14 @@ import com.alibaba.excel.EasyExcel;
import com.njcn.jbsyncdata.pojo.*;
import com.njcn.jbsyncdata.service.DisPhotovoltaicService;
import com.njcn.jbsyncdata.service.IBusinessService;
import com.njcn.jbsyncdata.util.RestTemplateUtil;
import com.njcn.jbsyncdata.util.StreamUtil;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
@@ -112,6 +114,45 @@ public class DisPhotovoltaicController {
}
}
/**
* @param date yyyy-MM-dd
*/
@ApiOperation(value = "查询指定日期所有台区用户的谐波遥测数据")
@PostMapping("/queryTelemetryAreaEmsData")
public void queryTelemetryAreaEmsData(String date) {
try {
businessService.queryTelemetryInterfaceAreaData(date);
} catch (Exception exception) {
exception.printStackTrace();
}
}
/**
* @param startTime 起始时间 yyyy-MM-dd
* @param endTime 截止时间 yyyy-MM-dd
*/
@ApiOperation(value = "查询日期范围内所有台区用户的谐波遥测数据")
@PostMapping("/queryTelemetryAreaEmsDataTimeRange")
public void queryTelemetryAreaEmsDataTimeRange(String startTime, String endTime) {
try {
DateTime startDate = DateUtil.parse(startTime, DatePattern.NORM_DATE_FORMAT);
DateTime endDate = DateUtil.parse(endTime, DatePattern.NORM_DATE_FORMAT);
long betweenDay = DateUtil.betweenDay(startDate, endDate, true);
//第一天
businessService.queryTelemetryInterfaceAreaData(DateUtil.format(startDate, DatePattern.NORM_DATE_PATTERN));
//后续递增1
for (int i = 0; i < betweenDay; i++) {
startDate = DateUtil.offsetDay(startDate, 1);
businessService.queryTelemetryInterfaceAreaData(DateUtil.format(startDate, DatePattern.NORM_DATE_PATTERN));
}
} catch (Exception exception) {
exception.printStackTrace();
}
}
@ApiOperation(value = "导入10kv分布式光伏接入情况", produces = MediaType.APPLICATION_OCTET_STREAM_VALUE)
@PostMapping("/import10KV")
public void import10(MultipartFile file, HttpServletResponse response) throws Exception {
@@ -205,7 +246,7 @@ public class DisPhotovoltaicController {
}
return "数据导入失败";
}
@Scheduled(cron = "0 30 8 * * ?")
@Scheduled(cron = "0 10 0 * * ?")
public void insert() {
log.error(Thread.currentThread().getName(),"1.定时器启动----！");
DateTime dateTime = DateUtil.offsetDay(new Date(), -1);
@@ -218,4 +259,36 @@ public class DisPhotovoltaicController {
log.error(Thread.currentThread().getName() + "2.定时器执行数据成功 "+ds+"----！");
}
@Scheduled(cron = "0 10 2 * * ?")
public void insertPms() {
log.error(Thread.currentThread().getName(),"1.定时器启动----！");
DateTime dateTime = DateUtil.offsetDay(new Date(), -1);
String s=dateTime.toString();
String ds = s.substring(0, s.indexOf(" "));
log.error(Thread.currentThread().getName() + "1.定时器执行遥测数据日期 "+ds+"----！");
businessService.queryTelemetryInterfaceAreaData(ds);
log.error(Thread.currentThread().getName() + "2.定时器执行数据成功 "+ds+"----！");
}
@ApiOperation(value = "将台区数据导入到配网表中")
@PostMapping("/cs")
public String importAreaaa() {
Map<String, String> headers=new HashMap<>();
headers.put("Content-Type", "application/xml");
String aa="<soapenv:Envelope xmlns:soapenv=\"http://schemas.xmlsoap.org/soap/envelope/\" xmlns:face=\"http://face.service.prss.com.app\">\n" +
" <soapenv:Header/>\n" +
" <soapenv:Body>\n" +
" <face:getBizcDytsYc>\n" +
" <face:in0>{\"data\":{\"pms_id\":\"634cb757ff8af6df8752399e910152634c8e4b3bfd\",\"startdate\":\"2024-06-10 00:00:00\",\"enddate\":\"2024-06-10 03:00:00\"}}</face:in0>\n" +
" </face:getBizcDytsYc>\n" +
" </soapenv:Body>\n" +
"</soapenv:Envelope>";
RestTemplateUtil restTemplateUtil = new RestTemplateUtil();
ResponseEntity<String> userEntity = restTemplateUtil.post("http://10.118.110.221:8013/prssface-ems/ws/EmsInterface",headers,aa, String.class);
userEntity.getBody();
System.out.println();
return userEntity.getBody().length() + "";
}
}

141
src/main/java/com/njcn/jbsyncdata/enums/EmsInterfaceTypeEnum.java Normal file
View File

@@ -0,0 +1,141 @@
package com.njcn.jbsyncdata.enums;
import lombok.Getter;
import java.util.ArrayList;
import java.util.List;
@Getter
public enum EmsInterfaceTypeEnum {
A_VTHD("70100","A相电压谐波总畸变率HVA","A","data_v","v_thd"),
A_V3("70103","A相3次电压谐波","A","data_harmrate_v","v_3"),
A_V4("70104","A相4次电压谐波","A","data_harmrate_v","v_4"),
A_V5("70105","A相5次电压谐波","A","data_harmrate_v","v_5"),
A_V6("70106","A相6次电压谐波","A","data_harmrate_v","v_6"),
A_V7("70107","A相7次电压谐波","A","data_harmrate_v","v_7"),
A_V8("70108","A相8次电压谐波","A","data_harmrate_v","v_8"),
A_V9("70109","A相9次电压谐波","A","data_harmrate_v","v_9"),
A_V10("70110","A相10次电压谐波","A","data_harmrate_v","v_10"),
A_V11("70111","A相11次电压谐波","A","data_harmrate_v","v_11"),
A_V12("70112","A相12次电压谐波","A","data_harmrate_v","v_12"),
A_V13("70113","A相13次电压谐波","A","data_harmrate_v","v_13"),
B_VTHD("70200","B相电压谐波总畸变率HVB","B","data_v","v_thd"),
B_V3("70203","B相3次电压谐波","B","data_harmrate_v","v_3"),
B_V4("70204","B相4次电压谐波","B","data_harmrate_v","v_4"),
B_V5("70205","B相5次电压谐波","B","data_harmrate_v","v_5"),
B_V6("70206","B相6次电压谐波","B","data_harmrate_v","v_6"),
B_V7("70207","B相7次电压谐波","B","data_harmrate_v","v_7"),
B_V8("70208","B相8次电压谐波","B","data_harmrate_v","v_8"),
B_V9("70209","B相9次电压谐波","B","data_harmrate_v","v_9"),
B_V10("70210","B相10次电压谐波","B","data_harmrate_v","v_10"),
B_V11("70211","B相11次电压谐波","B","data_harmrate_v","v_11"),
B_V12("70212","B相12次电压谐波","B","data_harmrate_v","v_12"),
B_V13("70213","B相13次电压谐波","B","data_harmrate_v","v_13"),
C_V4("70221","C相4次电压谐波","C","data_harmrate_v","v_4"),
C_V6("70223","C相6次电压谐波","C","data_harmrate_v","v_6"),
C_V8("70225","C相8次电压谐波","C","data_harmrate_v","v_8"),
C_V10("70227","C相10次电压谐波","C","data_harmrate_v","v_10"),
C_V12("70229","C相12次电压谐波","C","data_harmrate_v","v_12"),
A_I4("70231","A相4次电流谐波","A","data_i","i_4"),
A_I6("70232","A相6次电流谐波","A","data_i","i_6"),
A_I8("70233","A相8次电流谐波","A","data_i","i_8"),
A_I10("70234","A相10次电流谐波","A","data_i","i_10"),
A_I12("70235","A相12次电流谐波","A","data_i","i_12"),
B_I4("70236","B相4次电流谐波","B","data_i","i_4"),
B_I6("70237","B相6次电流谐波","B","data_i","i_6"),
B_I8("70238","B相8次电流谐波","B","data_i","i_8"),
B_I10("70239","B相10次电流谐波","B","data_i","i_10"),
B_I12("70240","B相12次电流谐波","B","data_i","i_12"),
C_I4("70241","C相4次电流谐波","C","data_i","i_4"),
C_I6("70242","C相6次电流谐波","C","data_i","i_6"),
C_I8("70243","C相8次电流谐波","C","data_i","i_8"),
C_I10("70244","C相10次电流谐波","C","data_i","i_10"),
C_I12("70245","C相12次电流谐波","C","data_i","i_12"),
A_V2("70246","A相2次电压谐波","A","data_harmrate_v","v_2"),
B_V2("70247","B相2次电压谐波","B","data_harmrate_v","v_2"),
C_V2("70248","C相2次电压谐波","C","data_harmrate_v","v_2"),
A_I2("70249","A相2次电流谐波","A","data_i","i_2"),
B_I2("70250","B相2次电流谐波","B","data_i","i_2"),
C_I2("70251","C相2次电流谐波","C","data_i","i_2"),
C_VTHD("70300","C相电压谐波总畸变率HVC","C","data_v","v_thd"),
C_V3("70303","C相3次电压谐波","C","data_harmrate_v","v_3"),
C_V5("70305","C相5次电压谐波","C","data_harmrate_v","v_5"),
C_V7("70307","C相7次电压谐波","C","data_harmrate_v","v_7"),
C_V9("70309","C相9次电压谐波","C","data_harmrate_v","v_9"),
C_V11("70311","C相11次电压谐波","C","data_harmrate_v","v_11"),
C_V13("70313","C相13次电压谐波","C","data_harmrate_v","v_13"),
A_I3("70910","A相3次电流谐波","A","data_i","i_3"),
A_I5("70911","A相5次电流谐波","A","data_i","i_5"),
A_I7("70912","A相7次电流谐波","A","data_i","i_7"),
A_I9("70913","A相9次电流谐波","A","data_i","i_9"),
A_I11("70914","A相11次电流谐波","A","data_i","i_11"),
A_I13("70915","A相13次电流谐波","A","data_i","i_13"),
B_I3("70919","B相3次电流谐波","B","data_i","i_3"),
B_I5("70920","B相5次电流谐波","B","data_i","i_5"),
B_I7("70921","B相7次电流谐波","B","data_i","i_7"),
B_I9("70922","B相9次电流谐波","B","data_i","i_9"),
B_I11("70923","B相11次电流谐波","B","data_i","i_11"),
B_I13("70924","B相13次电流谐波","B","data_i","i_13"),
C_I3("70928","C相3次电流谐波","C","data_i","i_3"),
C_I5("70929","C相5次电流谐波","C","data_i","i_5"),
C_I7("70930","C相7次电流谐波","C","data_i","i_7"),
C_I9("70931","C相9次电流谐波","C","data_i","i_9"),
C_I11("70932","C相11次电流谐波","C","data_i","i_11"),
C_I13("70933","C相13次电流谐波","C","data_i","i_13"),
A_ITHD("70952","A相电流谐波畸变率","A","data_i","i_thd"),
B_ITHD("70953","B相电流谐波畸变率","B","data_i","i_thd"),
C_ITHD("70954","C相电流谐波畸变率","C","data_i","i_thd"),
;
//冀北指标名称
private final String measType;
//中文名
private final String cnName;
//相别
private final String phaseType;
//表名
private final String tableName;
//字段名
private final String fieldName;
EmsInterfaceTypeEnum(String measType, String cnName, String phaseType, String tableName, String fieldName) {
this.measType = measType;
this.cnName = cnName;
this.phaseType = phaseType;
this.tableName = tableName;
this.fieldName = fieldName;
}
/**
* 根据指标获取当前指标的信息
* @param measType 指标名称
*/
public static EmsInterfaceTypeEnum getMeasTypeEnumByMeasType(String measType) {
for (EmsInterfaceTypeEnum item : values()) {
if (item.getMeasType().equals(measType)) {
return item;
}
}
return null;
}
/**
* 获取所有指标类型名称集合
*/
public static List<String> getMeasList() {
List<String> measLiST = new ArrayList<>();
for (EmsInterfaceTypeEnum temp : values()) {
String type = temp.getMeasType();
measLiST.add(type);
}
return measLiST;
}
}

4
src/main/java/com/njcn/jbsyncdata/mapper/PmsPowerDistributionareaMapper.java
View File

@@ -2,6 +2,9 @@ package com.njcn.jbsyncdata.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.njcn.jbsyncdata.pojo.po.PmsPowerDistributionarea;
import org.apache.ibatis.annotations.Param;
import java.util.List;
/**
* <p>
@@ -13,4 +16,5 @@ import com.njcn.jbsyncdata.pojo.po.PmsPowerDistributionarea;
*/
public interface PmsPowerDistributionareaMapper extends BaseMapper<PmsPowerDistributionarea> {
List<String> getDeptIds(@Param("id") String id);
}

4
src/main/java/com/njcn/jbsyncdata/mapper/mapping/PmsPowerDistributionareaMapper.xml
View File

@@ -2,4 +2,8 @@
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.njcn.jbsyncdata.mapper.PmsPowerDistributionareaMapper">
<select id="getDeptIds" resultType="java.lang.String">
select code from sys_dept where find_in_set(#{id},pids) and state = 1
</select>
</mapper>

6
src/main/java/com/njcn/jbsyncdata/pojo/DistributionAreaExcel.java
View File

@@ -31,7 +31,7 @@ public class DistributionAreaExcel implements Serializable {
/**
* 变压器名称
*/
@ExcelProperty("变压器名称")
@ExcelProperty("台区名称")
private String name;
/**
@@ -49,8 +49,8 @@ public class DistributionAreaExcel implements Serializable {
@ExcelProperty(value = "公司名称")
private String orgName;
@ExcelProperty(value = "供电所")
private String powerSupply;
@ExcelProperty(value = "变电站名称")
private String powerrName;
@ExcelProperty(value = "无数据的指标")
private String types;

2
src/main/java/com/njcn/jbsyncdata/pojo/po/DataV.java
View File

@@ -1,7 +1,7 @@
package com.njcn.jbsyncdata.pojo.po;
import com.fasterxml.jackson.databind.annotation.JsonSerialize;
import com.njcn.jbsyncdata.util.InstantDateSerializer;
import com.njcn.influx.utils.InstantDateSerializer;
import lombok.Data;
import org.influxdb.annotation.Column;
import org.influxdb.annotation.Measurement;

49
src/main/java/com/njcn/jbsyncdata/pojo/po/EmsInterface.java Normal file
View File

@@ -0,0 +1,49 @@
package com.njcn.jbsyncdata.pojo.po;
import lombok.Data;
/**
* @author wr
* @description
* @date 2024/6/11 15:46
*/
@Data
public class EmsInterface {
//主键
private String ID;
//市、县公司
private String SXGS;
//遥测时间
private String YCSJ;
//终端ID
private String ZDID;
//测点ID
private String CDID;
//设备ID
private String SBID;
//PMS2.0编号
private String PMSBH;
//测点值
private String CDZ;
//遥测类型
private String YCLX;
//更新时间
private String GXSJ;
//据插入时间
private String SJCRSJ;
//资源id
private String PMS_ID;
}

11
src/main/java/com/njcn/jbsyncdata/pojo/po/PmsPowerDistributionarea.java
View File

@@ -1,5 +1,6 @@
package com.njcn.jbsyncdata.pojo.po;
import com.alibaba.excel.annotation.ExcelProperty;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
@@ -26,18 +27,21 @@ public class PmsPowerDistributionarea {
* 台区编号
*/
@TableId("Id")
@ExcelProperty(value = "台区编号")
private String id;
/**
* 台区名称
*/
@TableField("Name")
@ExcelProperty("台区名称")
private String name;
/**
* 组织机构名称
*/
@TableField("Org_Name")
@ExcelProperty("所属部门")
private String orgName;
/**
@@ -62,6 +66,7 @@ public class PmsPowerDistributionarea {
* 变电站名称
*/
@TableField("Powerr_Name")
@ExcelProperty(value = "变电站名称")
private String powerrName;
/**
@@ -74,6 +79,7 @@ public class PmsPowerDistributionarea {
* 监测线路名称
*/
@TableField("Line_Name")
@ExcelProperty("所属线路")
private String lineName;
/**
@@ -209,6 +215,7 @@ public class PmsPowerDistributionarea {
private String updateBy;
@TableField("Pms_Id")
@ExcelProperty("PMS资源id（同源）")
private String pmsID;
/**
* 更新时间
@@ -216,5 +223,7 @@ public class PmsPowerDistributionarea {
@TableField("Update_Time")
private LocalDateTime updateTime;
@ExcelProperty(value = "无数据的指标")
@TableField(exist = false)
private String types;
}

6
src/main/java/com/njcn/jbsyncdata/service/IBusinessService.java
View File

@@ -8,5 +8,11 @@ public interface IBusinessService {
void queryTelemetryAreaData(String date);
/**
* 获取台区遥测数据
* @param date
*/
void queryTelemetryInterfaceAreaData(String date);
}

41
src/main/java/com/njcn/jbsyncdata/service/impl/BusinessServiceImpl.java
View File

@@ -9,12 +9,14 @@ import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.njcn.jbsyncdata.component.TokenComponent;
import com.njcn.jbsyncdata.enums.MeasTypeEnum;
import com.njcn.jbsyncdata.mapper.PmsPowerDistributionareaMapper;
import com.njcn.jbsyncdata.pojo.po.PmsPowerDistributionarea;
import com.njcn.jbsyncdata.pojo.result.*;
import com.njcn.jbsyncdata.service.IBusinessService;
import com.njcn.jbsyncdata.service.IPmsPowerDistributionareaService;
import com.njcn.jbsyncdata.service.IPmsPowerGenerationUserService;
import com.njcn.jbsyncdata.util.AreaDataProcessing;
import com.njcn.jbsyncdata.util.AreaInterfaceDataProcessing;
import com.njcn.jbsyncdata.util.DataProcessing;
import com.njcn.jbsyncdata.util.RestTemplateUtil;
import lombok.extern.slf4j.Slf4j;
@@ -39,12 +41,16 @@ public class BusinessServiceImpl implements IBusinessService {
@Resource
private AreaDataProcessing areaDataProcessing;
@Resource
private AreaInterfaceDataProcessing areaInterfaceDataProcessing;
@Resource
private IPmsPowerGenerationUserService pmsPowerGenerationUserService;
@Resource
private IPmsPowerDistributionareaService powerDistributionareaService;
@Resource
private PmsPowerDistributionareaMapper pmsPowerDistributionareaMapper;
/**
* 此方法通过发电客户编号查询数据，该方法存在以下问题
* 问题一：一个发电客户编号同指标返回的数据会有多个，但是目前看到最多2个测量点数据。
@@ -75,7 +81,7 @@ public class BusinessServiceImpl implements IBusinessService {
JSONObject jsonObject = JSONUtil.createObj();
JSONObject jsonObjectSub = JSONUtil.createObj();
jsonObject.set("page", 1);
jsonObject.set("perPage", 10000);
jsonObject.set("perPage", 2000);
jsonObject.set("startTime", DateUtil.format(beginOfDay, DatePattern.NORM_DATETIME_FORMATTER));
jsonObject.set("endTime", DateUtil.format(endOfDay, DatePattern.NORM_DATETIME_FORMATTER));
//1公专变2低压用户3光伏
@@ -93,7 +99,7 @@ public class BusinessServiceImpl implements IBusinessService {
List<String> userIds = pmsPowerGenerationUserService.queryAllUserId();
List<List<String>> singleQueryDataUserId = ListUtils.partition(userIds, 20000);
for (int k = 0; k < singleQueryDataUserId.size(); k++) {
dataProcessing.asyncInfluxDb(tokenComponent,date, restTemplateUtil, typeList, jsonObject, jsonObjectSub, headers, singleQueryDataUserId, k);
dataProcessing.asyncInfluxDb(tokenComponent, date, restTemplateUtil, typeList, jsonObject, jsonObjectSub, headers, singleQueryDataUserId, k);
}
}
@@ -113,7 +119,7 @@ public class BusinessServiceImpl implements IBusinessService {
JSONObject jsonObject = JSONUtil.createObj();
JSONObject jsonObjectSub = JSONUtil.createObj();
jsonObject.set("page", 1);
jsonObject.set("perPage", 10000);
jsonObject.set("perPage", 2000);
jsonObject.set("startTime", DateUtil.format(beginOfDay, DatePattern.NORM_DATETIME_FORMATTER));
jsonObject.set("endTime", DateUtil.format(endOfDay, DatePattern.NORM_DATETIME_FORMATTER));
//1公专变2低压用户3光伏
@@ -137,9 +143,34 @@ public class BusinessServiceImpl implements IBusinessService {
.collect(Collectors.toList());
List<List<PmsPowerDistributionarea>> singleQueryDataUserId = ListUtils.partition(list, 20000);
for (int k = 0; k < singleQueryDataUserId.size(); k++) {
areaDataProcessing.asyncInfluxDb(tokenComponent,date, restTemplateUtil, typeList, jsonObject, jsonObjectSub, headers, singleQueryDataUserId, k);
areaDataProcessing.asyncInfluxDb(tokenComponent, date, restTemplateUtil, typeList, jsonObject, jsonObjectSub, headers, singleQueryDataUserId, k);
}
}
@Override
public void queryTelemetryInterfaceAreaData(String date) {
DateTime dateTemp = DateUtil.parse(date, DatePattern.NORM_DATE_FORMAT);
DateTime beginOfDay = DateUtil.beginOfDay(dateTemp);
DateTime endOfDay = DateUtil.endOfDay(dateTemp);
String startTime = DateUtil.format(beginOfDay, DatePattern.NORM_DATETIME_FORMATTER);
String endTime = DateUtil.format(endOfDay, DatePattern.NORM_DATETIME_FORMATTER);
List<String> deptIds = pmsPowerDistributionareaMapper.getDeptIds("2adc64baf2308725c4e91105b3186b21");
//获取所有发电用户的id
List<PmsPowerDistributionarea> list = powerDistributionareaService.list(new LambdaQueryWrapper<PmsPowerDistributionarea>()
.select(PmsPowerDistributionarea::getId, PmsPowerDistributionarea::getPmsID)
.in(PmsPowerDistributionarea::getOrgId, deptIds)
.isNotNull(PmsPowerDistributionarea::getPmsID)
);
list = list.stream()
.filter(t -> StrUtil.isNotBlank(t.getPmsID()))
.collect(Collectors.toList());
List<List<PmsPowerDistributionarea>> singleQueryDataUserId = ListUtils.partition(list, 2000);
for (int k = 0; k < singleQueryDataUserId.size(); k++) {
areaInterfaceDataProcessing.asyncInfluxDb(startTime, endTime, singleQueryDataUserId.get(k), k);
}
System.out.println("结束");
}
}

6
src/main/java/com/njcn/jbsyncdata/service/impl/DisPhotovoltaicServiceImpl.java
View File

@@ -279,16 +279,16 @@ public class DisPhotovoltaicServiceImpl implements DisPhotovoltaicService {
area.setOperationName(dept.get(0).getName());
area.setOperationId(dept.get(0).getCode());
}
if(StrUtil.isNotEmpty(excel.getPowerSupply())){
if(StrUtil.isNotEmpty(excel.getPowerrName())){
List<PmsStatationStat> station = oldList.stream()
.filter(x ->x.getPowerName().contains(excel.getPowerSupply()))
.filter(x ->x.getPowerName().contains(excel.getPowerrName()))
.collect(Collectors.toList());
if (CollUtil.isNotEmpty(station)){
area.setPowerrName(station.get(0).getPowerName());
area.setPowerStationId(station.get(0).getPowerId());
area.setVoltageLevel(station.get(0).getVoltageLevel());
}else {
area.setPowerrName(excel.getPowerSupply());
area.setPowerrName(excel.getPowerrName());
area.setPowerStationId(IdUtil.simpleUUID());
area.setVoltageLevel(dictData.getId());
}

2
src/main/java/com/njcn/jbsyncdata/util/AreaDataProcessing.java
View File

@@ -57,7 +57,7 @@ public class AreaDataProcessing {
) {
TokenResult tokenWithRestTemplate;
//将发电用户编号按100尺寸分片
List<List<PmsPowerDistributionarea>> partitionList = ListUtils.partition(singleQueryDataUserId.get(k), 100);
List<List<PmsPowerDistributionarea>> partitionList = ListUtils.partition(singleQueryDataUserId.get(k), 20);
log.error("总计分了{}片", partitionList.size());
int count = 0;
tokenWithRestTemplate = tokenComponent.getTokenWithRestTemplate();

272
src/main/java/com/njcn/jbsyncdata/util/AreaInterfaceDataProcessing.java Normal file
View File

@@ -0,0 +1,272 @@
package com.njcn.jbsyncdata.util;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.file.FileWriter;
import cn.hutool.core.text.StrPool;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.njcn.influx.utils.InfluxDbUtils;
import com.njcn.jbsyncdata.enums.EmsInterfaceTypeEnum;
import com.njcn.jbsyncdata.pojo.InfluxAreaData;
import com.njcn.jbsyncdata.pojo.po.EmsInterface;
import com.njcn.jbsyncdata.pojo.po.PmsPowerDistributionarea;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.ListUtils;
import org.apache.commons.lang3.StringUtils;
import org.influxdb.InfluxDB;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.springframework.http.ResponseEntity;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.xml.sax.InputSource;
import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.DocumentBuilderFactory;
import java.io.File;
import java.io.StringReader;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
/**
* @author wr
* @description
* @date 2023/10/20 14:14
*/
@Component
@RequiredArgsConstructor
@Slf4j
public class AreaInterfaceDataProcessing {
private final InfluxDbUtils influxDbUtils;
@Async("asyncExecutor")
public void asyncInfluxDb(
String startTime,
String endTime,
List<PmsPowerDistributionarea> singleQueryDataUserId, int k
) {
StringReader stringReader;
InputSource inputSource;
DocumentBuilder documentBuilder;
Document document;
int count = 0;
//获取初始化数据信息
List<EmsInterface> responseEntities = new ArrayList<>();
// // 创建 DocumentBuilder 对象
// DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
// DocumentBuilder builder;
// try {
// builder = factory.newDocumentBuilder();
// // 解析 SOAP XML 文件
// Document document = builder.parse("C:\\Users\\web2023\\Desktop\\response.xml");
// // 获取 SOAP Body 元素
// Element body = (Element) document.getElementsByTagName("soap:Body").item(0);
// // 获取 GetStockPrice 元素
// Element stockPrice = (Element) body.getElementsByTagName("ns1:getBizcDytsYcResponse").item(0);
// // 获取 StockName 元素的文本内容
// String stockName = stockPrice.getElementsByTagName("ns1:out").item(0).getTextContent();
// Map ma = JSONObject.parseObject(stockName, Map.class);
// responseEntities.addAll(JSONArray.parseArray(JSONArray.toJSONString(ma.get("data")), EmsInterface.class));
// } catch (ParserConfigurationException | SAXException | IOException e) {
// throw new RuntimeException(e);
// }
//没有遥测数据的信息
List<String> notEmsInterface = new ArrayList<>();
int kk = k + 1;
for (PmsPowerDistributionarea areaIds : singleQueryDataUserId) {
count++;
//log.error("查询第{}大片，{}小片数据", kk, count);
try {
Map<String, String> headers = new HashMap<>();
headers.put("Content-Type", "application/xml");
String aa = "<soapenv:Envelope xmlns:soapenv=\"http://schemas.xmlsoap.org/soap/envelope/\" xmlns:face=\"http://face.service.prss.com.app\">\n" +
" <soapenv:Header/>\n" +
" <soapenv:Body>\n" +
" <face:getBizcDytsYc>\n" +
" <face:in0>{\"data\":{\"pms_id\":\"" + areaIds.getPmsID() + "\",\"startdate\":\"" + startTime + "\",\"enddate\":\"" + endTime + "\"}}</face:in0>\n" +
" </face:getBizcDytsYc>\n" +
" </soapenv:Body>\n" +
"</soapenv:Envelope>";
RestTemplateUtil restTemplateUtil = new RestTemplateUtil();
ResponseEntity<String> userEntity = restTemplateUtil.post("http://10.118.110.221:8013/prssface-ems/ws/EmsInterface", headers, aa, String.class);
if (userEntity.getStatusCodeValue() == 200) {
String entityBody = userEntity.getBody();
stringReader = new StringReader(entityBody);
inputSource = new InputSource(stringReader);
documentBuilder = DocumentBuilderFactory.newInstance().newDocumentBuilder();
document = documentBuilder.parse(inputSource);
// 获取 SOAP Body 元素
Element body = (Element) document.getElementsByTagName("soap:Body").item(0);
// 获取 GetStockPrice 元素
Element stockPrice = (Element) body.getElementsByTagName("ns1:getBizcDytsYcResponse").item(0);
// 获取 StockName 元素的文本内容
String stockName = stockPrice.getElementsByTagName("ns1:out").item(0).getTextContent();
Map ma = JSONObject.parseObject(stockName, Map.class);
if(ma.containsKey("data")){
List<EmsInterface> resul = JSONArray.parseArray(com.alibaba.fastjson.JSONArray.toJSONString(ma.get("data")), EmsInterface.class);
if(CollUtil.isNotEmpty(resul)){
responseEntities.addAll(resul);
}else{
notEmsInterface.add(areaIds.getPmsID());
}
}
}
} catch (Exception exception) {
//log.error("远程调用接口异常，异常为：" + exception);
}
}
List<String> typeList = EmsInterfaceTypeEnum.getMeasList();
//开始解析数据
Set<String> userIdConcatMeasType = new HashSet<>();
//将指标+pms_id组合起来匹配返回数据的第一条记录:pms_Id@measType
for (String measType : typeList) {
userIdConcatMeasType.addAll(singleQueryDataUserId.stream().map(x -> x.getPmsID().concat(StrPool.AT).concat(measType)).collect(Collectors.toSet()));
}
List</*各值以逗号分隔*/InfluxAreaData> influxData;
Map</*表名*/String, List</*各值以逗号分隔*/String>> typeData = new HashMap<>();
StringBuilder tempInfluxData;
String dataIdentify;
EmsInterfaceTypeEnum measTypeEnumByMeasType;
//获取资源id和台区的对应关系
Map<String, List<String>> areaMap = singleQueryDataUserId.stream().collect(Collectors.groupingBy(PmsPowerDistributionarea::getPmsID, Collectors.mapping(PmsPowerDistributionarea::getId, Collectors.toList())));
//数据处理
for (EmsInterface emsInterface : responseEntities) {
dataIdentify = emsInterface.getPMS_ID().concat(StrPool.AT).concat(emsInterface.getYCLX());
if (userIdConcatMeasType.contains(dataIdentify)) {
//获取该类型枚举
measTypeEnumByMeasType = EmsInterfaceTypeEnum.getMeasTypeEnumByMeasType(emsInterface.getYCLX());
influxData = new ArrayList<>();
InfluxAreaData influxAreaData;
if(areaMap.containsKey(emsInterface.getPMS_ID())){
List<String> list = areaMap.get(emsInterface.getPMS_ID());
for (String s : list) {
tempInfluxData = new StringBuilder();
tempInfluxData.append(measTypeEnumByMeasType.getPhaseType())
.append(StrPool.COMMA)
.append(s)
.append(StrPool.COMMA)
.append(emsInterface.getYCSJ())
.append(StrPool.COMMA)
.append(measTypeEnumByMeasType.getFieldName())
.append(StrPool.COMMA)
.append(StrUtil.isBlank(emsInterface.getCDZ()) ? "0" : emsInterface.getCDZ());
influxAreaData=new InfluxAreaData();
influxAreaData.setId(s);
influxAreaData.setInfluxData(tempInfluxData.toString());
influxData.add(influxAreaData);
}
//userId@measType@tableName:存在多个指标存储表名一致，避免数据覆盖；
Map<String, List<String>> influxLineData = influxData.stream().collect(Collectors.groupingBy(InfluxAreaData::getId, Collectors.mapping(InfluxAreaData::getInfluxData, Collectors.toList())));
for (String s : influxLineData.keySet()) {
typeData.put(s.concat(StrPool.AT).concat(measTypeEnumByMeasType.getMeasType()).concat(StrPool.AT).concat(measTypeEnumByMeasType.getTableName()), influxLineData.get(s));
}
//处理完，删除该条记录，减少集合尺寸，提高效率
userIdConcatMeasType.remove(dataIdentify);
}
} else {
//log.error("查询遥测数据失败！结果为：{}", dataIdentify);
}
}
//没有匹配上的就是该用户没有数据
//log.error("剩余有{}条标识", userIdConcatMeasType.size());
//最后输出没有数据的资源编号
/**
* 输出到2个文件，lackData.txt、 excalationData.txt
* 注：
* 1、所有指标均没有有数据的资源编号
* 2、部分指标没有数据的资源编号，并表明是哪些指标
*/
if (CollectionUtil.isNotEmpty(userIdConcatMeasType)) {
Map<String, List<String>> finalMap = userIdConcatMeasType.stream().collect(Collectors.groupingBy(str ->
str.substring(0, str.indexOf(StrPool.AT))
));
//全部缺失数据的用户
List<String> lackData = new ArrayList<>();
//部分缺失的用户及指标
List<String> excalationData = new ArrayList<>();
Set<String> keyedSet = finalMap.keySet();
for (String key : keyedSet) {
List<String> data = finalMap.get(key);
if (data.size() == typeList.size()) {
lackData.add(key);
} else {
data = data.stream().map(t -> t.substring(t.indexOf(StrPool.AT) + 1)).collect(Collectors.toList());
key = key.concat(StrPool.COMMA).concat(StringUtils.join(data, StrPool.AT));
excalationData.add(key);
}
}
FileWriter lackDataWriter = FileWriter.create(new File("/usr/local/syncDataEms/lackData" + startTime.substring(0,10) + k + "台区.txt"));
lackDataWriter.writeLines(lackData);
FileWriter excalationDataWriter = FileWriter.create(new File("/usr/local/syncDataEms/excalationData" + startTime.substring(0,10) + k + "台区.txt"));
excalationDataWriter.writeLines(excalationData);
}
//log.error("用户有指标没有数据的长度为：{}", userIdConcatMeasType.size());
//最后批量入库
batchInsertData(typeData);
}
/**
* 批量入库influxDB
*
* @param typeData 远程根据用户编号获取的数据 Map</表名/String, List<Map</属性名/String,/数值/String>>> typeData = new HashMap<>();
*/
private void batchInsertData(Map<String, List<String>> typeData) {
//log.error("总计有{}条记录入库，以20000作为基数分片插入influxdb", typeData.size());
List<String> sqlList = new ArrayList<>();
Set<String> tableNames = typeData.keySet();
String[] datas;
Map<String, String> tags;
Map<String, Object> fields;
Point point;
BatchPoints batchPoints;
for (String tableName : tableNames) {
List<String> data = typeData.get(tableName);
tableName = tableName.substring(tableName.lastIndexOf(StrPool.AT) + 1);
for (String datum : data) {
datas = datum.split(StrPool.COMMA);
//tag数据
tags = new HashMap<>();
tags.put("phasic_type", datas[0]);
tags.put("line_id", datas[1]);
tags.put("quality_flag", "0");
tags.put("value_type", "AVG");
String time = datas[2];
//tag数据删完后，剩余均是filed数据,因filed属性名不固定，无法指定获取，直接循环
fields = new HashMap<>();
fields.put(datas[3], datas[4]);
point = influxDbUtils.pointBuilder(tableName, DateUtil.parse(time, DatePattern.NORM_DATETIME_MS_PATTERN).getTime(), TimeUnit.MILLISECONDS, tags, fields);
batchPoints = BatchPoints.database(influxDbUtils.getDbName()).retentionPolicy("").consistency(InfluxDB.ConsistencyLevel.ALL).build();
batchPoints.point(point);
sqlList.add(batchPoints.lineProtocol());
}
}
List<List<String>> subSqlList = ListUtils.partition(sqlList, 20000);
int count = 1;
for (List<String> sql : subSqlList) {
try {
influxDbUtils.batchInsert(influxDbUtils.getDbName(), "autogen", InfluxDB.ConsistencyLevel.ALL, TimeUnit.MILLISECONDS, sql);
} catch (Exception exception) {
//log.error("数据批量入库异常，异常为：{}", exception.toString());
exception.printStackTrace();
}
log.error("已经入库{}条记录!", count * 20000);
count++;
}
log.error("当前批次所有数据，{}条均已入库！", sqlList.size());
}
}

2
src/main/java/com/njcn/jbsyncdata/util/DataProcessing.java
View File

@@ -54,7 +54,7 @@ public class DataProcessing {
) {
TokenResult tokenWithRestTemplate;
//将发电用户编号按100尺寸分片
List<List<String>> partitionList = ListUtils.partition(singleQueryDataUserId.get(k), 100);
List<List<String>> partitionList = ListUtils.partition(singleQueryDataUserId.get(k), 20);
log.error("总计分了{}片", partitionList.size());
int count = 0;
tokenWithRestTemplate = tokenComponent.getTokenWithRestTemplate();

38
src/main/java/com/njcn/jbsyncdata/util/InstantDateDeserializer.java
View File

@@ -1,38 +0,0 @@
package com.njcn.jbsyncdata.util;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.DeserializationContext;
import com.fasterxml.jackson.databind.deser.std.StdDeserializer;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.time.Instant;
/**
* @author hongawen
* @version 1.0.0
* @date 2023年07月24日 13:33
*/
@Component
public class InstantDateDeserializer extends StdDeserializer<Instant> {
public InstantDateDeserializer() {
this(null);
}
protected InstantDateDeserializer(Class<?> vc) {
super(vc);
}
@Override
public Instant deserialize(JsonParser p, DeserializationContext ctxt) throws IOException, JsonProcessingException {
String text = p.getValueAsString();
return PubUtils.dateToInstant(DateUtil.parse(text,DatePattern.NORM_DATETIME_PATTERN));
}
}

48
src/main/java/com/njcn/jbsyncdata/util/InstantDateSerializer.java
View File

@@ -1,48 +0,0 @@
package com.njcn.jbsyncdata.util;
import cn.hutool.core.date.DatePattern;
import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.databind.SerializerProvider;
import com.fasterxml.jackson.databind.ser.std.StdSerializer;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.time.Instant;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
/**
* @author hongawen
* @version 1.0.0
* @date 2023年04月25日 16:33
*/
@Component
public class InstantDateSerializer extends StdSerializer<Instant> {
private static final long serialVersionUID = 1L;
private static DateTimeFormatter format = DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN);
public InstantDateSerializer() {
this(null);
}
public InstantDateSerializer(Class<Instant> t) {
super(t);
}
/***
* 转义为 yyyy-MM-dd HH:mm:ss
* @author hongawen
*/
@Override
public void serialize(Instant instant, JsonGenerator jsonGenerator, SerializerProvider provider)
throws IOException {
if (instant == null) {
return;
}
String jsonValue = format.format(instant.atZone(ZoneId.of("+00:00")));
jsonGenerator.writeString(jsonValue);
}
}