组态功能开发

cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/controller/ledger/CsLedgerController.java

@@ -49,6 +49,15 @@ public class CsLedgerController extends BaseController {
         return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, list, methodDescribe);
     }
 
+    @OperateInfo(info = LogEnum.BUSINESS_COMMON)
+    @PostMapping("/lineTree")
+    @ApiOperation("监测点树")
+    public HttpResult<List<CsLedgerVO>> lineTree(){
+        String methodDescribe = getMethodDescribe("lineTree");
+        List<CsLedgerVO> list = csLedgerService.getLineTree();
+        return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, list, methodDescribe);
+    }
+
     @OperateInfo(info = LogEnum.BUSINESS_COMMON)
     @PostMapping("/add")
     @ApiOperation("新增台账信息")

cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/service/ICsLedgerService.java

@@ -21,6 +21,12 @@ public interface ICsLedgerService {
      */
     List<CsLedgerVO> getLedgerTree();
 
+    /**
+     * 获取终端台账树
+     * @return
+     */
+    List<CsLedgerVO> getLineTree();
+
     /**
      * 新增台账数据
      * @param csLedgerParam

cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/service/impl/CsLedgerServiceImpl.java

@@ -42,6 +42,20 @@ public class CsLedgerServiceImpl extends ServiceImpl<CsLedgerMapper, CsLedger> i
     @Override
     public List<CsLedgerVO> getLedgerTree() {
         List<CsLedgerVO> list = new ArrayList<>();
+        List<CsLedgerVO> engineeringList = getLineTree();
+        CsLedgerVO vo = new CsLedgerVO();
+        vo.setId("9999999");
+        vo.setLevel(0);
+        vo.setName("台账管理");
+        if (CollectionUtil.isNotEmpty(engineeringList)) {
+            vo.setChildren(engineeringList);
+        }
+        list.add(vo);
+        return list;
+    }
+
+    @Override
+    public List<CsLedgerVO> getLineTree() {
         List<CsLedgerVO> engineeringList = new ArrayList<>();
         List<CsLedgerVO> allList = this.baseMapper.getAll();
         //fixme 这边先根据登录的用户名称来区分是否展示所有的台账信息
@@ -57,15 +71,7 @@ public class CsLedgerServiceImpl extends ServiceImpl<CsLedgerMapper, CsLedger> i
         deviceList.forEach(dev -> dev.setChildren(getChildren(dev, lineList)));
         projectList.forEach(pro -> pro.setChildren(getChildren(pro, deviceList)));
         engineeringList.forEach(eng -> eng.setChildren(getChildren(eng, projectList)));
-        CsLedgerVO vo = new CsLedgerVO();
+        return engineeringList;
-        vo.setId("9999999");
-        vo.setLevel(0);
-        vo.setName("台账管理");
-        if (CollectionUtil.isNotEmpty(engineeringList)) {
-            vo.setChildren(engineeringList);
-        }
-        list.add(vo);
-        return list;
     }
 
     @Override

cs-harmonic/cs-harmonic-api/src/main/java/com/njcn/harmonic/constant/HarmonicConstant.java

@@ -20,4 +20,56 @@ public interface HarmonicConstant {
      * 组态Json文件oss文件名
      */
     String CONFIGURATIONNAME = "configuration.json";
+
+    String DATA_V = "data_v";
+
+    String DATA_I = "data_i";
+
+    String DATA_HARM_RATE_V = "data_harmrate_v";
+
+    String DATA_HARM_RATE_I = "data_harmrate_i";
+
+    String DATA_HARM_PHASIC_V = "data_harmphasic_v";
+
+    String DATA_HARM_PHASIC_I = "data_harmphasic_i";
+
+    String DATA_IN_HARM_V = "data_inharm_v";
+
+    String DATA_IN_HARM_I = "data_inharm_i";
+
+    String DATA_HARM_POWER_P = "data_harmpower_p";
+
+    String DATA_HARM_POWER_Q = "data_harmpower_q";
+
+    String DATA_HARM_POWER_S = "data_harmpower_s";
+
+    String DATA_FLICKER = "data_flicker";
+
+    String DATA_FLUC = "data_fluc";
+
+    String DATA_PLT = "data_plt";
+
+    String EPD_DATA = "epd_data";
+
+    String PQD_DATA = "pqd_data";
+
+    String BMD_DATA = "bmd_data";
+
+    String EVT_DATA = "evt_data";
+
+    String ALM_DATA = "alm_data";
+
+    String STS_DATA = "sts_data";
+
+    String DI_DATA = "di_data";
+
+    String DO_DATA = "do_data";
+
+    String PARM_DATA = "parm_data";
+
+    String SET_DATA = "set_data";
+
+    String INSET_DATA = "inset_data";
+
+    String CTRL_DATA = "ctrl_data";
 }

cs-harmonic/cs-harmonic-api/src/main/java/com/njcn/harmonic/pojo/influx/CsCommon.java

@@ -0,0 +1,36 @@
+package com.njcn.harmonic.pojo.influx;
+
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.njcn.common.utils.serializer.InstantDateSerializer;
+import lombok.Data;
+import org.influxdb.annotation.Column;
+
+import java.time.Instant;
+
+/**
+ * 类的介绍：influx实时数据通用实体
+ *
+ * @author xuyang
+ * @version 1.0.0
+ * @createTime 2023/6/6 9:26
+ */
+@Data
+public class CsCommon {
+
+    @Column(name = "time")
+    @JsonSerialize(using = InstantDateSerializer.class)
+    private Instant time;
+
+    @Column(name = "line_id")
+    private String lineId;
+
+    @Column(name = "phasic_type")
+    private String phaseType;
+
+    @Column(name = "value_type")
+    private String valueType;
+
+    @Column(name = "rtData")
+    private Double rtData;
+
+}

cs-harmonic/cs-harmonic-api/src/main/java/com/njcn/harmonic/pojo/influx/CsDataHarmPhasicI.java

@@ -0,0 +1,18 @@
+package com.njcn.harmonic.pojo.influx;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import org.influxdb.annotation.Measurement;
+
+/**
+ * 类的介绍：
+ *
+ * @author xuyang
+ * @version 1.0.0
+ * @createTime 2023/6/6 9:27
+ */
+@EqualsAndHashCode(callSuper = true)
+@Data
+@Measurement(name = "data_harmphasic_i")
+public class CsDataHarmPhasicI extends CsCommon{
+}

cs-harmonic/cs-harmonic-api/src/main/java/com/njcn/harmonic/pojo/influx/CsDataHarmPhasicV.java

@@ -0,0 +1,18 @@
+package com.njcn.harmonic.pojo.influx;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import org.influxdb.annotation.Measurement;
+
+/**
+ * 类的介绍：
+ *
+ * @author xuyang
+ * @version 1.0.0
+ * @createTime 2023/6/6 9:27
+ */
+@EqualsAndHashCode(callSuper = true)
+@Data
+@Measurement(name = "data_harmphasic_v")
+public class CsDataHarmPhasicV extends CsCommon{
+}

cs-harmonic/cs-harmonic-api/src/main/java/com/njcn/harmonic/pojo/influx/CsDataHarmPowerP.java

@@ -0,0 +1,17 @@
+package com.njcn.harmonic.pojo.influx;
+
+import lombok.Data;
+import org.influxdb.annotation.Measurement;
+
+/**
+ * 类的介绍：
+ *
+ * @author xuyang
+ * @version 1.0.0
+ * @createTime 2023/6/6 9:05
+ */
+@Data
+@Measurement(name = "data_harmpower_p")
+public class CsDataHarmPowerP extends CsCommon {
+
+}

cs-harmonic/cs-harmonic-api/src/main/java/com/njcn/harmonic/pojo/influx/CsDataHarmRateI.java

@@ -0,0 +1,18 @@
+package com.njcn.harmonic.pojo.influx;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import org.influxdb.annotation.Measurement;
+
+/**
+ * 类的介绍：
+ *
+ * @author xuyang
+ * @version 1.0.0
+ * @createTime 2023/6/6 9:27
+ */
+@EqualsAndHashCode(callSuper = true)
+@Data
+@Measurement(name = "data_harmrate_i")
+public class CsDataHarmRateI extends CsCommon{
+}

cs-harmonic/cs-harmonic-api/src/main/java/com/njcn/harmonic/pojo/influx/CsDataHarmRateV.java

@@ -0,0 +1,18 @@
+package com.njcn.harmonic.pojo.influx;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import org.influxdb.annotation.Measurement;
+
+/**
+ * 类的介绍：
+ *
+ * @author xuyang
+ * @version 1.0.0
+ * @createTime 2023/6/6 9:27
+ */
+@EqualsAndHashCode(callSuper = true)
+@Data
+@Measurement(name = "data_harmrate_v")
+public class CsDataHarmRateV extends CsCommon{
+}

cs-harmonic/cs-harmonic-api/src/main/java/com/njcn/harmonic/pojo/influx/CsDataI.java

@@ -0,0 +1,18 @@
+package com.njcn.harmonic.pojo.influx;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import org.influxdb.annotation.Measurement;
+
+/**
+ * 类的介绍：
+ *
+ * @author xuyang
+ * @version 1.0.0
+ * @createTime 2023/6/6 9:27
+ */
+@EqualsAndHashCode(callSuper = true)
+@Data
+@Measurement(name = "data_i")
+public class CsDataI extends CsCommon{
+}

cs-harmonic/cs-harmonic-api/src/main/java/com/njcn/harmonic/pojo/influx/CsDataV.java

@@ -0,0 +1,18 @@
+package com.njcn.harmonic.pojo.influx;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import org.influxdb.annotation.Measurement;
+
+/**
+ * 类的介绍：
+ *
+ * @author xuyang
+ * @version 1.0.0
+ * @createTime 2023/6/6 9:27
+ */
+@EqualsAndHashCode(callSuper = true)
+@Data
+@Measurement(name = "data_v")
+public class CsDataV extends CsCommon{
+}

cs-harmonic/cs-harmonic-api/src/main/java/com/njcn/harmonic/pojo/influx/CsPqdData.java

@@ -0,0 +1,18 @@
+package com.njcn.harmonic.pojo.influx;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import org.influxdb.annotation.Measurement;
+
+/**
+ * 类的介绍：
+ *
+ * @author xuyang
+ * @version 1.0.0
+ * @createTime 2023/6/6 11:21
+ */
+@EqualsAndHashCode(callSuper = true)
+@Data
+@Measurement(name = "pqd_data")
+public class CsPqdData extends CsCommon{
+}

cs-harmonic/cs-harmonic-api/src/main/java/com/njcn/harmonic/pojo/vo/ElementsVO.java

@@ -19,23 +19,32 @@ public class ElementsVO implements Serializable {
     @SerializedName("type")
     private String type;
 
-    @SerializedName("pointId")
+    @SerializedName("lineId")
-    private String pointId;
+    private List<String> lineId;
 
-    @SerializedName("pointName")
+    @SerializedName("lineName")
-    private String pointName;
+    private String lineName;
 
     @SerializedName("targetId")
-    private String targetId;
+    private List<String> targetId;
+
+    @SerializedName("targetPid")
+    private String targetPid;
 
     @SerializedName("targetName")
     private String targetName;
 
+    @SerializedName("dataType")
+    private String dataType;
+
+    @SerializedName("phasic")
+    private String phasic;
+
     @SerializedName("defaultPic")
     private String defaultPic;
 
     @SerializedName("width")
-    private Integer width;
+    private String width;
 
     @SerializedName("height")
     private Integer height;
@@ -49,4 +58,10 @@ public class ElementsVO implements Serializable {
     @SerializedName("condition")
     private List<ConditionVO> condition;
 
+    @SerializedName("color")
+    private String color;
+
+    @SerializedName("fontSize")
+    private Integer fontSize;
+
 }

cs-harmonic/cs-harmonic-api/src/main/java/com/njcn/harmonic/pojo/vo/RtDataVO.java

@@ -0,0 +1,18 @@
+package com.njcn.harmonic.pojo.vo;
+
+import com.njcn.harmonic.pojo.influx.CsCommon;
+import lombok.Data;
+
+/**
+ * 类的介绍：
+ *
+ * @author xuyang
+ * @version 1.0.0
+ * @createTime 2023/6/6 11:38
+ */
+@Data
+public class RtDataVO extends CsCommon {
+
+    private String targetName;
+
+}

cs-harmonic/cs-harmonic-api/src/main/java/com/njcn/harmonic/utils/CsReflectUitl.java

@@ -0,0 +1,81 @@
+package com.njcn.harmonic.utils;
+
+import org.influxdb.annotation.Measurement;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URL;
+import java.net.URLDecoder;
+import java.util.ArrayList;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.List;
+
+/**
+ * Description:
+ * Date: 2023/6/2  15:08【需求编号】
+ *
+ * @author clam
+ * @version V1.0.0
+ */
+public class CsReflectUitl {
+
+
+    /*
+    * 获取po报下，所有有@Measurement注解的实体类;
+    */
+    public static HashMap<String,Class<?>> getEntityClassesByAnnotation() {
+        HashMap<String,Class<?>> result = new HashMap<>();
+        List<Class<?>> classes = CsReflectUitl.getClasses("com.njcn.harmonic.pojo.influx");
+        for (Class<?> clazz : classes) {
+            if (clazz.isAnnotationPresent(Measurement.class)) {
+                Measurement annotation = (Measurement) clazz.getAnnotation(Measurement.class);
+                result.put(annotation.name(),clazz) ;
+            }
+        }
+        return result;
+    }
+
+    public static List<Class<?>> getClasses(String packageName) {
+        List<Class<?>> classes = new ArrayList<Class<?>>();
+        try {
+            String path = packageName.replace(".", "/");
+            Enumeration<URL> resources = Thread.currentThread().getContextClassLoader().getResources(path);
+            while (resources.hasMoreElements()) {
+                URL resource = resources.nextElement();
+                if (resource.getProtocol().equals("file")) {
+                    String filePath = URLDecoder.decode(resource.getFile(), "UTF-8");
+                    findClassesInPackage(packageName, filePath, classes);
+                }
+            }
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+        return classes;
+    }
+
+    private static void findClassesInPackage(String packageName, String packagePath, List<Class<?>> classes) {
+        File packageDir = new File(packagePath);
+        if (!packageDir.exists() || !packageDir.isDirectory()) {
+            return;
+        }
+        File[] files = packageDir.listFiles();
+        for (File file : files) {
+            if (file.isDirectory()) {
+                findClassesInPackage(packageName + "." + file.getName(), file.getAbsolutePath(), classes);
+            } else if (file.getName().endsWith(".class")) {
+                String className = packageName + "." + file.getName().substring(0, file.getName().length() - 6);
+                try {
+                    Class<?> clazz = Class.forName(className);
+                    classes.add(clazz);
+                } catch (ClassNotFoundException e) {
+                    e.printStackTrace();
+                }
+            }
+        }
+    }
+
+
+
+
+}

cs-harmonic/cs-harmonic-boot/pom.xml

@@ -74,6 +74,18 @@
             <version>1.0.0</version>
             <scope>compile</scope>
         </dependency>
+        <dependency>
+            <groupId>com.njcn</groupId>
+            <artifactId>common-influxDB</artifactId>
+            <version>${project.version}</version>
+            <scope>compile</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.njcn</groupId>
+            <artifactId>system-api</artifactId>
+            <version>1.0.0</version>
+            <scope>compile</scope>
+        </dependency>
     </dependencies>
 
     <build>

cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/harmonic/controller/LineTargetController.java

@@ -6,6 +6,7 @@ import com.njcn.common.pojo.enums.response.CommonResponseEnum;
 import com.njcn.common.pojo.response.HttpResult;
 import com.njcn.common.utils.HttpResultUtil;
 import com.njcn.csdevice.pojo.vo.LineTargetVO;
+import com.njcn.harmonic.pojo.vo.RtDataVO;
 import com.njcn.harmonic.pojo.vo.TargetDetailVO;
 import com.njcn.harmonic.service.ILineTargetService;
 import com.njcn.web.controller.BaseController;
@@ -65,10 +66,11 @@ public class LineTargetController extends BaseController {
     @PostMapping("/lineData")
     @ApiOperation("获取指标数据")
     @ApiImplicitParam(name = "id", value = "组态页面id", required = true)
-    public HttpResult<String> getLineData(@RequestParam("id") String id) {
+    @Deprecated
+    public HttpResult<List<RtDataVO>> getLineData(@RequestParam("id") String id) {
         String methodDescribe = getMethodDescribe("getLineData");
-        lineTargetService.getLineData(id);
+        List<RtDataVO> list = lineTargetService.getLineData(id);
-        return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
+        return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, list, methodDescribe);
     }
 
 }

cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/harmonic/handler/MqttMessageHandler.java

@@ -0,0 +1,40 @@
+package com.njcn.harmonic.handler;
+
+import com.alibaba.nacos.shaded.com.google.gson.Gson;
+import com.github.tocrhz.mqtt.annotation.MqttSubscribe;
+import com.github.tocrhz.mqtt.annotation.NamedValue;
+import com.github.tocrhz.mqtt.annotation.Payload;
+import com.github.tocrhz.mqtt.publisher.MqttPublisher;
+import com.njcn.harmonic.pojo.vo.RtDataVO;
+import com.njcn.harmonic.service.ILineTargetService;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+
+/**
+ * @author hongawen
+ * @version 1.0.0
+ * @date 2022年03月23日 09:41
+ */
+@Slf4j
+@Component
+@AllArgsConstructor
+public class MqttMessageHandler {
+
+    private final MqttPublisher publisher;
+
+    private final ILineTargetService lineTargetService;
+
+    /**
+     * 实时数据应答
+     */
+    @MqttSubscribe(value = "/zl/askRtData/{pageId}",qos = 1)
+    public void responseRtData(String topic, @NamedValue("pageId") String pageId, MqttMessage message, @Payload String payload) {
+        List<RtDataVO> list = lineTargetService.getLineData(pageId);
+        Gson gson = new Gson();
+        publisher.send("/zl/rtData/"+pageId,gson.toJson(list),1,false);
+    }
+}

cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/harmonic/service/ILineTargetService.java

@@ -1,6 +1,7 @@
 package com.njcn.harmonic.service;
 
 import com.njcn.csdevice.pojo.vo.LineTargetVO;
+import com.njcn.harmonic.pojo.vo.RtDataVO;
 import com.njcn.harmonic.pojo.vo.TargetDetailVO;
 
 import java.util.List;
@@ -29,7 +30,6 @@ public interface ILineTargetService {
      * 获取绑定指标的数据
      * @param id
      */
-    void getLineData(String id);
+    List<RtDataVO> getLineData(String id);
-
 
 }

cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/harmonic/service/impl/LineTargetServiceImpl.java

@@ -1,18 +1,31 @@
 package com.njcn.harmonic.service.impl;
 
+import cn.hutool.core.collection.CollectionUtil;
 import com.alibaba.nacos.shaded.com.google.gson.Gson;
+import com.github.tocrhz.mqtt.publisher.MqttPublisher;
 import com.njcn.csdevice.api.*;
 import com.njcn.csdevice.pojo.po.CsDataArray;
 import com.njcn.csdevice.pojo.po.CsDevModelPO;
 import com.njcn.csdevice.pojo.po.CsDevModelRelationPO;
 import com.njcn.csdevice.pojo.vo.LineTargetVO;
+import com.njcn.harmonic.constant.HarmonicConstant;
+import com.njcn.harmonic.pojo.influx.CsDataHarmPowerP;
+import com.njcn.harmonic.pojo.influx.CsDataV;
+import com.njcn.harmonic.pojo.influx.CsPqdData;
 import com.njcn.harmonic.pojo.vo.ElementsVO;
+import com.njcn.harmonic.pojo.vo.RtDataVO;
 import com.njcn.harmonic.pojo.vo.TargetDetailVO;
 import com.njcn.harmonic.pojo.vo.ZuTaiVo;
 import com.njcn.harmonic.service.CsPagePOService;
 import com.njcn.harmonic.service.ILineTargetService;
+import com.njcn.harmonic.utils.CsReflectUitl;
+import com.njcn.influxdb.utils.InfluxDbUtils;
 import com.njcn.oss.utils.FileStorageUtil;
+import com.njcn.system.api.EpdFeignClient;
 import lombok.AllArgsConstructor;
+import org.influxdb.dto.QueryResult;
+import org.influxdb.impl.InfluxDBResultMapper;
+import org.springframework.beans.BeanUtils;
 import org.springframework.stereotype.Service;
 
 import java.io.BufferedReader;
@@ -20,6 +33,7 @@ import java.io.InputStream;
 import java.io.InputStreamReader;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Objects;
 import java.util.stream.Collectors;
@@ -49,6 +63,13 @@ public class LineTargetServiceImpl implements ILineTargetService {
 
     private final CsPagePOService csPagePOService;
 
+    private final MqttPublisher publisher;
+
+    private final InfluxDbUtils influxDbUtils;
+
+    private final EpdFeignClient epdFeignClient;
+
+
     @Override
     public List<LineTargetVO> getLineTarget(String lineId) {
         List<LineTargetVO> list = new ArrayList<>();
@@ -77,12 +98,26 @@ public class LineTargetServiceImpl implements ILineTargetService {
     }
 
     @Override
-    public void getLineData(String id) {
+    public List<RtDataVO> getLineData(String id) {
+        List<RtDataVO> result = new ArrayList<>();
         String path = csPagePOService.queryById(id).getPath();
         InputStream inputStream = fileStorageUtil.getFileStream(path);
         List<ElementsVO> list = analysisJson(inputStream);
-        System.out.println("list==:" + list);
+        //influx的原始方法获取数据
-
+        if (CollectionUtil.isNotEmpty(list)){
+            list.forEach(item->{
+                String targetName = item.getTargetId().stream().reduce((first, second) -> second).orElse("no last element");
+                if (CollectionUtil.isNotEmpty(dataArrayFeignClient.getDataArrayById(item.getTargetPid(),targetName).getData())){
+                    String dataId = dataArrayFeignClient.getDataArrayById(item.getTargetPid(),targetName).getData().get(0).getDataId();
+                    String classId = epdFeignClient.selectById(dataId).getData().getClassId();
+                    String lineId = item.getLineId().stream().reduce((first, second) -> second).orElse("no last element");
+                    String dataType = item.getDataType();
+                    String phasic = item.getPhasic();
+                    result.add(getLineRtData2(lineId,classId,targetName,phasic,dataType,targetName));
+                }
+            });
+        }
+        return result;
     }
 
     /**
@@ -105,4 +140,78 @@ public class LineTargetServiceImpl implements ILineTargetService {
         return zuTai.getElements();
     }
 
+    /**
+     * 通过拼接sql语句获取方法
+     * @param lineId        监测点Id
+     * @param tableName     表名称
+     * @param columnName    字段名称
+     * @param phasic        相别
+     * @param dataType      数据类型
+     * @return
+     */
+    public <T> T getLineRtData(String lineId, String tableName, String columnName, String phasic, String dataType) {
+        T t = null;
+        HashMap<String, Class<?>> entityClassesByAnnotation = CsReflectUitl.getEntityClassesByAnnotation();
+        Class<?> aClass = entityClassesByAnnotation.get(tableName);
+        String sql = "select line_id,phasic_type,value_type,last(" + columnName + ") AS rtData from "+ tableName +" where line_id = '" + lineId + "' and phasic_type = '" + phasic + "' and value_type = '" + dataType + "' tz('Asia/Shanghai')";
+        QueryResult sqlResult = influxDbUtils.query(sql);
+        InfluxDBResultMapper resultMapper = new InfluxDBResultMapper();
+        List<?> list = resultMapper.toPOJO(sqlResult, aClass);
+        if (CollectionUtil.isNotEmpty(list)){
+            t = (T)list.get(0);
+        }
+        return t;
+    }
+
+    /**
+     * 通过拼接sql语句获取方法
+     * @param lineId        监测点Id
+     * @param tableName     表名称
+     * @param columnName    字段名称
+     * @param phasic        相别
+     * @param dataType      数据类型
+     * @return
+     */
+    public RtDataVO getLineRtData2(String lineId, String tableName, String columnName, String phasic, String dataType, String target) {
+        RtDataVO vo = new RtDataVO();
+        String sql = "select line_id,phasic_type,value_type,last(" + columnName + ") AS rtData from "+ tableName +" where line_id = '" + lineId + "' and phasic_type = '" + phasic + "' and value_type = '" + dataType + "' tz('Asia/Shanghai')";
+        QueryResult sqlResult = influxDbUtils.query(sql);
+        if (CollectionUtil.isNotEmpty(sqlResult.getResults())){
+            vo.setRtData(Double.parseDouble(sqlResult.getResults().get(0).getSeries().get(0).getValues().get(0).get(4).toString()));
+        }
+        vo.setLineId(lineId);
+        vo.setPhaseType(phasic);
+        vo.setValueType(dataType);
+        vo.setValueType(dataType);
+        vo.setTargetName(target);
+        return vo;
+    }
+
+    /**
+     * 根据表明返回不同数据结果
+     */
+    public RtDataVO findDataByTableName(String lineId, String tableName, String columnName, String phasic, String dataType,String targetName) {
+        RtDataVO rtDataVO = new RtDataVO();
+        switch (tableName) {
+            case HarmonicConstant.DATA_HARM_POWER_P:
+                CsDataHarmPowerP rtData1 = getLineRtData(lineId,tableName,targetName,phasic,dataType);
+                BeanUtils.copyProperties(rtData1,rtDataVO);
+                rtDataVO.setTargetName(targetName);
+                break;
+            case HarmonicConstant.DATA_V:
+                CsDataV rtData2 = getLineRtData(lineId,tableName,targetName,phasic,dataType);
+                BeanUtils.copyProperties(rtData2,rtDataVO);
+                rtDataVO.setTargetName(targetName);
+                break;
+            case HarmonicConstant.PQD_DATA:
+                CsPqdData rtData3 = getLineRtData(lineId,tableName,targetName,phasic,dataType);
+                BeanUtils.copyProperties(rtData3,rtDataVO);
+                rtDataVO.setTargetName(targetName);
+                break;
+            default:
+                break;
+        }
+        return rtDataVO;
+    }
+
 }

cs-harmonic/cs-harmonic-boot/src/test/java/com/njcn/InfluxDbTest.java

@@ -0,0 +1,61 @@
+package com.njcn;
+
+import com.alibaba.nacos.shaded.com.google.gson.Gson;
+import com.github.tocrhz.mqtt.publisher.MqttPublisher;
+import com.njcn.harmonic.pojo.vo.RtDataVO;
+import com.njcn.harmonic.service.ILineTargetService;
+import com.njcn.influxdb.param.InfluxDBPublicParam;
+import com.njcn.influxdb.utils.InfluxDbUtils;
+import lombok.AllArgsConstructor;
+import org.influxdb.InfluxDB;
+import org.influxdb.dto.BatchPoints;
+import org.influxdb.dto.Point;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import javax.annotation.Resource;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 类的介绍：
+ *
+ * @author xuyang
+ * @version 1.0.0
+ * @createTime 2023/6/5 14:44
+ */
+@AllArgsConstructor
+public class InfluxDbTest {
+
+    private final MqttPublisher publisher;
+
+    public static void main(String[] args) {
+
+
+    }
+
+
+    public void insert() {
+        InfluxDbUtils influxDbUtils = new InfluxDbUtils("admin", "123456", "http://192.168.1.16:8086", "pqsbase_zl", "");
+        List<String> records = new ArrayList<>();
+        List<String> phasic = Arrays.asList("A","B","C");
+        List<String> dataType = Arrays.asList("max","min","avg","cp95");
+        long time = System.currentTimeMillis();
+        for (String item1 : phasic) {
+            for (String item2 : dataType) {
+                Map<String, String> tags = new HashMap<>();
+                Map<String, Object> fields = new HashMap<>();
+                tags.put("line_id","4");
+                tags.put("phasic_type",item1);
+                tags.put("value_type",item2);
+                fields.put("W",new Random().nextDouble());
+                fields.put("PhV",new Random().nextDouble());
+                Point point = influxDbUtils.pointBuilder("pqd_data", time, TimeUnit.MILLISECONDS, tags, fields);
+                BatchPoints batchPoints = BatchPoints.database(influxDbUtils.getDbName ()).tag(InfluxDBPublicParam.LINE_ID, "4").tag(InfluxDBPublicParam.PHASIC_TYPE,item1).tag(InfluxDBPublicParam.VALUE_TYPE,item2).retentionPolicy("").consistency(InfluxDB.ConsistencyLevel.ALL).build();
+                batchPoints.point(point);
+                records.add(batchPoints.lineProtocol());
+            }
+        }
+        influxDbUtils.batchInsert(influxDbUtils.getDbName (),"", InfluxDB.ConsistencyLevel.ALL, records);
+    }
+
+}