9 changed files with 628 additions and 109 deletions
@ -0,0 +1,19 @@ |
|||||
|
package com.techsor.datacenter.sender.entitiy.kingio; |
||||
|
|
||||
|
|
||||
|
import lombok.Data; |
||||
|
|
||||
|
import java.util.List; |
||||
|
|
||||
|
@Data |
||||
|
public class TrendLog1Entity { |
||||
|
private String Instance; |
||||
|
private List<LogBuffer> log_buffer; |
||||
|
|
||||
|
@Data |
||||
|
public static class LogBuffer { |
||||
|
private String timestamp; |
||||
|
private String logData; |
||||
|
private String DataType; |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,34 @@ |
|||||
|
package com.techsor.datacenter.sender.entitiy.kingio; |
||||
|
|
||||
|
|
||||
|
import lombok.Data; |
||||
|
|
||||
|
import java.util.List; |
||||
|
|
||||
|
import com.fasterxml.jackson.annotation.JsonProperty; |
||||
|
import lombok.Data; |
||||
|
|
||||
|
@Data |
||||
|
public class TrendLog1RawEntity { |
||||
|
|
||||
|
@JsonProperty("Instance") |
||||
|
private String Instance; |
||||
|
|
||||
|
@JsonProperty("DeviceId") |
||||
|
private String DeviceId; |
||||
|
|
||||
|
@JsonProperty("log_buffer") |
||||
|
private List<LogBuffer> log_buffer; |
||||
|
|
||||
|
@Data |
||||
|
public static class LogBuffer { |
||||
|
@JsonProperty("timestamp") |
||||
|
private String timestamp; |
||||
|
|
||||
|
@JsonProperty("logData") |
||||
|
private String logData; |
||||
|
|
||||
|
@JsonProperty("DataType") |
||||
|
private String DataType; |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,251 @@ |
|||||
|
package com.techsor.datacenter.sender.service; |
||||
|
|
||||
|
import com.alibaba.fastjson2.JSON; |
||||
|
import com.fasterxml.jackson.databind.ObjectMapper; |
||||
|
import com.google.gson.Gson; |
||||
|
import com.techsor.datacenter.sender.components.CommonOpt; |
||||
|
import com.techsor.datacenter.sender.config.DataSourceContextHolder; |
||||
|
import com.techsor.datacenter.sender.dao.KingIOServerDAO; |
||||
|
import com.techsor.datacenter.sender.dao.TypeDAO; |
||||
|
import com.techsor.datacenter.sender.entitiy.DeviceEntity; |
||||
|
import com.techsor.datacenter.sender.entitiy.kingio.*; |
||||
|
import com.techsor.datacenter.sender.utils.TimeUtils; |
||||
|
import jakarta.annotation.Resource; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
|
||||
|
import java.io.IOException; |
||||
|
import java.text.DecimalFormat; |
||||
|
import java.text.ParseException; |
||||
|
import java.text.SimpleDateFormat; |
||||
|
import java.time.LocalDateTime; |
||||
|
import java.time.format.DateTimeFormatter; |
||||
|
import java.time.format.DateTimeParseException; |
||||
|
import java.util.ArrayList; |
||||
|
import java.util.Date; |
||||
|
import java.util.List; |
||||
|
import java.util.Map; |
||||
|
|
||||
|
/** |
||||
|
* 服务类，用于处理TrendLog的数据。 |
||||
|
* <p> |
||||
|
* 此服务类负责接收和解析TrendLog的数据，并根据数据处理逻辑生成相应的数据消息。 |
||||
|
*/ |
||||
|
@Slf4j |
||||
|
@Component("TrendLogService") |
||||
|
public class TrendLogService { |
||||
|
|
||||
|
@Resource |
||||
|
TypeDAO typeDAO; |
||||
|
@Resource |
||||
|
KingIOServerDAO kingIOServerDAO; |
||||
|
@Autowired |
||||
|
private CommonOpt commonOpt; |
||||
|
|
||||
|
private final ObjectMapper objectMapper = new ObjectMapper(); |
||||
|
private SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS"); |
||||
|
private static final DecimalFormat DECIMAL_FORMAT = new DecimalFormat("#0.00"); |
||||
|
/** |
||||
|
* 将JSON字符串解析为KingIODataModel对象。 |
||||
|
* |
||||
|
* @param json King IO Server发送的数据的JSON字符串 |
||||
|
* @return 解析后的KingIODataModel对象 |
||||
|
* @throws IOException 当解析JSON时发生错误 |
||||
|
*/ |
||||
|
public TrendLog1RawEntity parseJson(String json) throws IOException { |
||||
|
return objectMapper.readValue(json, TrendLog1RawEntity.class); |
||||
|
} |
||||
|
/** |
||||
|
* 开始处理King IO Server发来的数据。 |
||||
|
* <p> |
||||
|
* 该方法首先调用parseJson方法解析数据，然后根据解析后的数据对象进行进一步处理。 |
||||
|
* |
||||
|
* @param data King IO Server发送的数据的JSON字符串 |
||||
|
* @return 处理后的数据消息，如果无法处理则返回空字符串 |
||||
|
* @throws Exception 当解析JSON或处理数据过程中发生错误 |
||||
|
*/ |
||||
|
public String start(String data) throws Exception { |
||||
|
TrendLog1RawEntity dataModel = parseJson(data); |
||||
|
return processData(dataModel); |
||||
|
} |
||||
|
/** |
||||
|
* 根据解析后的KingIODataModel对象处理数据。 |
||||
|
* <p> |
||||
|
* 此方法遍历数据对象中的数据项，根据数据项的类型和内容生成相应的数据消息字符串。 |
||||
|
* |
||||
|
* @param data 解析后的KingIODataModel对象 |
||||
|
* @return 生成的数据消息字符串，如果无法处理则返回空字符串 |
||||
|
* @throws ParseException 当处理数据时发生错误 |
||||
|
*/ |
||||
|
public String processData(TrendLog1RawEntity data) throws ParseException { |
||||
|
log.info("TrendLog DataSize to process:{}",data.getLog_buffer().size()); |
||||
|
List<KingIODbmEntity> dbmValues = new ArrayList<>(); |
||||
|
for(TrendLog1RawEntity.LogBuffer dataItem : data.getLog_buffer()){ |
||||
|
String topCompanyId= commonOpt.getTopCompanyId(data.getDeviceId()); |
||||
|
if (topCompanyId==null || topCompanyId.equals("0")){ |
||||
|
continue; |
||||
|
} |
||||
|
//Switch database
|
||||
|
DataSourceContextHolder.setCurrentDataSourceKey("dataSourceForCompany_"+topCompanyId); |
||||
|
log.info("Use datasource for company:"+topCompanyId); |
||||
|
|
||||
|
KingIODbmEntity tempEntity = new KingIODbmEntity(); |
||||
|
DeviceEntity deviceItem = kingIOServerDAO.queryDeviceWsClientIdByDeviceId(data.getDeviceId()); |
||||
|
if (deviceItem==null){ |
||||
|
continue; |
||||
|
} |
||||
|
log.info("TrendLog process item:{},{},{}",data.getDeviceId(),dataItem.getLogData(),dataItem.getTimestamp()); |
||||
|
tempEntity.setDeviceId(deviceItem.getDeviceId()); |
||||
|
tempEntity.setContent(generateDBMStr(deviceItem,dataItem.getLogData())); |
||||
|
try { |
||||
|
tempEntity.setTs(TimeUtils.kingiOServerTimeToTs(dataItem.getTimestamp())); |
||||
|
}catch (Exception e){ |
||||
|
log.error("Error KingIOServer Data: KingIODataItemEntity timestamp to utc timestamp error"); |
||||
|
log.error("Received Time : "+dataItem.getTimestamp()); |
||||
|
tempEntity.setTs(""); |
||||
|
} |
||||
|
dbmValues.add(tempEntity); |
||||
|
DataSourceContextHolder.clearCurrentDataSourceKey(); |
||||
|
} |
||||
|
|
||||
|
return JSON.toJSONString(dbmValues); |
||||
|
} |
||||
|
|
||||
|
//处理TrendLog的单条数据
|
||||
|
public Boolean analyzeLogBufferData(Map<String, Object> obj,KingIODataModel data,List<KingIODataItemEntity> dataValues){ |
||||
|
String name = (String) obj.get("N"); |
||||
|
Object value = obj.getOrDefault("2", data.getPropertyValues().get("1")); |
||||
|
Object qualityStamp = obj.getOrDefault("3", data.getPropertyValues().get("3")); |
||||
|
String formattedTimestamp = (String) data.getPropertyValues().get("1"); |
||||
|
if (formattedTimestamp == null || formattedTimestamp.trim().isEmpty()) { |
||||
|
throw new IllegalArgumentException("formattedTimestamp is null or empty!"); |
||||
|
} |
||||
|
// 清理不可见字符并去除空格
|
||||
|
formattedTimestamp = formattedTimestamp.replaceAll("\\p{C}", "").trim(); |
||||
|
|
||||
|
try{ |
||||
|
// 检查质量戳
|
||||
|
//0=BAD
|
||||
|
//192=正常
|
||||
|
//65535=初期値&未収集
|
||||
|
// if (!qualityStamp.equals(192) && !qualityStamp.equals(65535) && !qualityStamp.equals(0)) {
|
||||
|
if (!qualityStamp.equals(192)) { |
||||
|
log.error("Error KingIOServer Data: 设备名 " + name + ", 质量戳: " + qualityStamp); |
||||
|
return false; |
||||
|
} |
||||
|
|
||||
|
try{ |
||||
|
// 处理时间戳
|
||||
|
if (obj.containsKey("2")) { |
||||
|
//以下注释是因为这里不要计算时间差，统一用数据头里的时间
|
||||
|
// long timestampOffset = Long.parseLong(obj.get("2").toString());
|
||||
|
Date date = parseDate(formattedTimestamp); |
||||
|
// date.setTime(date.getTime() + timestampOffset);
|
||||
|
formattedTimestamp = dateFormat.format(date); |
||||
|
} |
||||
|
}catch (Exception e){ |
||||
|
log.error("Error processing timeproperties:"+(String) data.getPropertyValues().get("2")); |
||||
|
return false; |
||||
|
} |
||||
|
|
||||
|
//Process the number value. keep 4 ecimal places
|
||||
|
if (value instanceof Number) { |
||||
|
DecimalFormat df = new DecimalFormat("#.####"); // 保留四位小数的格式
|
||||
|
String formatted = df.format((Number)value); |
||||
|
value = Double.parseDouble(formatted); |
||||
|
}else if (value instanceof Boolean){ |
||||
|
if((Boolean)value==true){ |
||||
|
value = 1; |
||||
|
}else{ |
||||
|
value = 0; |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
// 输出数据解读
|
||||
|
log.debug("DeviceName：{}" , name); |
||||
|
log.debug("Value：{}" , value); |
||||
|
log.debug("Time：{}" , formattedTimestamp); |
||||
|
|
||||
|
dataValues.add(new KingIODataItemEntity(name, value, formattedTimestamp, String.valueOf(qualityStamp))); |
||||
|
}catch (Exception e){ |
||||
|
log.error("DeviceName：{}" , name); |
||||
|
log.error("Value：{}" , value); |
||||
|
log.error("Time：{}" , formattedTimestamp); |
||||
|
log.error("KINGIoserver error:",e); |
||||
|
} |
||||
|
return Boolean.TRUE; |
||||
|
} |
||||
|
|
||||
|
|
||||
|
|
||||
|
/** |
||||
|
* 根据设备实体和数据项生成DBM消息字符串。 |
||||
|
* <p> |
||||
|
* 此方法根据设备的类型ID和数据项的内容生成特定格式的DBM消息字符串。 |
||||
|
* |
||||
|
* @param deviceItem 设备实体对象 |
||||
|
* @return 生成的DBM消息字符串 |
||||
|
*/ |
||||
|
private String generateDBMStr(DeviceEntity deviceItem,String value){ |
||||
|
String finalStr = ""; |
||||
|
String tempKey =""; |
||||
|
finalStr = String.format("{\"%s"+tempKey+"\":%s",deviceItem.getDeviceSN(),value+"}"); |
||||
|
return finalStr; |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* Transform boolan obj to int. If it's int, return itself. |
||||
|
* @param value |
||||
|
* @return |
||||
|
*/ |
||||
|
private Integer BooleanObjToInt(Object value){ |
||||
|
Integer tempValue; |
||||
|
|
||||
|
if (value instanceof Boolean) { |
||||
|
tempValue = (Boolean) value ? 1 : 0; // 布尔型处理
|
||||
|
} else if (value instanceof Integer) { |
||||
|
tempValue = (Integer) value; // 整型处理
|
||||
|
} else if (value instanceof Double) { |
||||
|
tempValue = Integer.valueOf(((Double) value).intValue()); |
||||
|
} |
||||
|
else { |
||||
|
throw new IllegalArgumentException("Unsupported value type: " + value.getClass()); |
||||
|
} |
||||
|
|
||||
|
return tempValue; |
||||
|
} |
||||
|
|
||||
|
public static Date parseDate(String timestamp) { |
||||
|
String[] patterns = { |
||||
|
"yyyy-MM-dd HH:mm:ss.SSS", // 3 位毫秒
|
||||
|
"yyyy-MM-dd HH:mm:ss.SSSS" // 4 位毫秒
|
||||
|
}; |
||||
|
|
||||
|
for (String pattern : patterns) { |
||||
|
try { |
||||
|
SimpleDateFormat tempDateFormat = new SimpleDateFormat(pattern); |
||||
|
return tempDateFormat.parse(timestamp); |
||||
|
} catch (Exception ignored) { } |
||||
|
} |
||||
|
throw new IllegalArgumentException("Invalid date format: " + timestamp); |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* 判断字符串是否为指定的日期格式：yyyy-M-d H:m:s |
||||
|
* 例如：2025-8-28 15:36:28 |
||||
|
*/ |
||||
|
public static boolean isDateString(String dateString) { |
||||
|
if (dateString == null || dateString.trim().isEmpty()) { |
||||
|
return false; |
||||
|
} |
||||
|
|
||||
|
try { |
||||
|
DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-M-d H:m:s"); |
||||
|
LocalDateTime.parse(dateString.trim(), formatter); |
||||
|
return true; |
||||
|
} catch (DateTimeParseException e) { |
||||
|
return false; |
||||
|
} |
||||
|
} |
||||
|
} |
||||
Loading…
Reference in new issue