package com.ruoyi.other.udan.service;
|
|
import com.alibaba.fastjson.JSON;
|
import com.google.protobuf.CodedInputStream;
|
import com.google.protobuf.WireFormat;
|
import com.ruoyi.other.udan.model.UdanEmsDeviceState;
|
import com.ruoyi.other.udan.util.UdanEmsKafkaUtil;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
import org.springframework.stereotype.Component;
|
|
import java.io.IOException;
|
|
/** 将 iems-open-v1.0.5.proto 消息解析为业务对象，并映射为待入库对象。 */
|
@Slf4j
|
@Component
|
@ConditionalOnProperty(prefix = "udan.ems.kafka", name = "enabled", havingValue = "true")
|
public class UdanEmsMessageProcessor {
|
private final UdanEmsKafkaUtil kafkaUtil;
|
|
/** 注入 Kafka 消费工具。 */
|
public UdanEmsMessageProcessor(UdanEmsKafkaUtil kafkaUtil) { this.kafkaUtil = kafkaUtil; }
|
|
/** 注册消息回调，避免消费者只消费但没有业务处理。 */
|
//@PostConstruct
|
public void register() { kafkaUtil.setMessageHandler(this::process); }
|
|
/** 解析消息、转换待入库对象并输出日志；此处暂不调用数据库。 */
|
public void process(String emsSystemId, byte[] protobufBytes) {
|
try {
|
UdanEmsDeviceState state = parse(emsSystemId, protobufBytes);
|
log.info("2、UDAN EMS实际对象: {}", JSON.toJSONString(state));
|
} catch (IOException e) {
|
log.error("UDAN EMS protobuf解析失败: emsSystemId={}, bytes={}", emsSystemId, protobufBytes == null ? 0 : protobufBytes.length, e);
|
}
|
}
|
|
/** 按顶层 IEmsAllDeviceState 字段号读取并构造真实业务对象。 */
|
public UdanEmsDeviceState parse(String id, byte[] bytes) throws IOException {
|
UdanEmsDeviceState s = new UdanEmsDeviceState();
|
s.setEmsSystemId(id);
|
CodedInputStream in = CodedInputStream.newInstance(bytes == null ? new byte[0] : bytes);
|
while (!in.isAtEnd()) {
|
int tag = in.readTag(); if (tag == 0) break;
|
int field = WireFormat.getTagFieldNumber(tag); int type = WireFormat.getTagWireType(tag);
|
if (field == 1 && type == WireFormat.WIRETYPE_VARINT) s.setTime(in.readUInt64());
|
else if (type == WireFormat.WIRETYPE_LENGTH_DELIMITED) parseChild(s, field, in.readBytes().toByteArray());
|
else in.skipField(tag);
|
}
|
return s;
|
}
|
|
|
/** 解析顶层各子消息，字段号来自 iems-open-v1.0.5.proto。 */
|
private void parseChild(UdanEmsDeviceState s, int field, byte[] data) throws IOException {
|
switch (field) {
|
case 2: s.setEms(parseEms(data)); break;
|
case 4: s.getPcsStates().add(parsePcs(data)); break;
|
case 5: s.getBatteryStackStates().add(parseStack(data)); break;
|
case 6: s.getBatteryClusterStates().add(parseCluster(data)); break;
|
case 7: s.getElectricMeterStates().add(parseMeter(data)); break;
|
case 8: s.setPowerMeterCount(s.getPowerMeterCount() + 1); break;
|
case 9: s.setAirConditionerCount(s.getAirConditionerCount() + 1); break;
|
case 11: s.setFireCount(s.getFireCount() + 1); break;
|
case 13: s.setJpfCount(s.getJpfCount() + 1); break;
|
case 14: s.setTmsCount(s.getTmsCount() + 1); break;
|
case 18: s.setDehumidifierCount(s.getDehumidifierCount() + 1); break;
|
case 19: s.setUpsCount(s.getUpsCount() + 1); break;
|
case 20: s.getPvStates().add(parsePv(data)); break;
|
case 21: s.getChargeStationStates().add(parseChargeStation(data)); break;
|
case 23: s.setExtPeripheralCount(s.getExtPeripheralCount() + 1); break;
|
default: break;
|
}
|
}
|
|
/** 解析 EMS 状态对象中的实际字段。 */
|
private UdanEmsDeviceState.EmsState parseEms(byte[] b) throws IOException {
|
UdanEmsDeviceState.EmsState x = new UdanEmsDeviceState.EmsState();
|
read(b, (f, t, in) -> { if (t == 2 && f == 1) x.setDeviceId(in.readString()); else if (t == 5 && f == 18) x.setDemandPower(in.readFloat()); else if (t == 2 && f == 21) x.setProtoProtocolVersion(in.readString()); else if (t == 5 && f == 33) x.setDailyChargeEnergyKwh(in.readFloat()); else if (t == 5 && f == 34) x.setDailyDischargeEnergyKwh(in.readFloat()); else if (t == 1 && f == 37) x.setTotalChargeEnergyKwh(in.readDouble()); else if (t == 1 && f == 38) x.setTotalDischargeEnergyKwh(in.readDouble()); else if (t == 5 && f == 43) x.setSumsSoc(in.readFloat()); else if (t == 5 && f == 69) x.setSumsSoh(in.readFloat()); else if (t == 5 && f == 41) x.setDcBusVoltage(in.readFloat()); else if (t == 5 && f == 42) x.setDcBusCurrent(in.readFloat()); else skip(in, t); });
|
return x;
|
}
|
|
/** 解析 PCS 实际字段。 */
|
private UdanEmsDeviceState.PcsState parsePcs(byte[] b) throws IOException {
|
UdanEmsDeviceState.PcsState x = new UdanEmsDeviceState.PcsState();
|
read(b, (f, t, in) -> { if (t == 2 && f == 2) x.setDeviceId(in.readString()); else if (t == 5 && f == 13) x.setAcTotalActivePower(in.readFloat()); else if (t == 5 && f == 21) x.setAcTotalReactivePower(in.readFloat()); else if (t == 5 && f == 22) x.setAcTotalApparentPower(in.readFloat()); else if (t == 5 && f == 23) x.setAcTotalPowerFactor(in.readFloat()); else if (t == 5 && f == 17) x.setDailyChargeEnergyKwh(in.readFloat()); else if (t == 5 && f == 18) x.setDailyDischargeEnergyKwh(in.readFloat()); else if (t == 1 && f == 19) x.setTotalChargeEnergyKwh(in.readDouble()); else if (t == 1 && f == 20) x.setTotalDischargeEnergyKwh(in.readDouble()); else if (t == 5 && f == 59) x.setAmbientTemp(in.readFloat()); else skip(in, t); });
|
return x;
|
}
|
|
/** 解析电池堆实际字段。 */
|
private UdanEmsDeviceState.BatteryStackState parseStack(byte[] b) throws IOException { UdanEmsDeviceState.BatteryStackState x = new UdanEmsDeviceState.BatteryStackState(); read(b, (f,t,in)-> { if(t==2&&f==2)x.setDeviceId(in.readString()); else if(t==5&&f==7)x.setTv(in.readFloat()); else if(t==5&&f==8)x.setTc(in.readFloat()); else if(t==5&&f==9)x.setSoh(in.readFloat()); else if(t==5&&f==10)x.setSoc(in.readFloat()); else if(t==5&&f==31)x.setRechargeableCapacityKwh(in.readFloat()); else if(t==5&&f==32)x.setDischargeableCapacityKwh(in.readFloat()); else if(t==1&&f==33)x.setCumulativeDischargeKwh(in.readDouble()); else if(t==1&&f==34)x.setCumulativeChargeKwh(in.readDouble()); else skip(in,t); }); return x; }
|
/** 解析电池簇实际字段。 */
|
private UdanEmsDeviceState.BatteryClusterState parseCluster(byte[] b) throws IOException { UdanEmsDeviceState.BatteryClusterState x = new UdanEmsDeviceState.BatteryClusterState(); read(b,(f,t,in)->{if(t==2&&f==2)x.setDeviceId(in.readString());else if(t==2&&f==3)x.setStackDeviceId(in.readString());else if(t==5&&f==7)x.setTv(in.readFloat());else if(t==5&&f==8)x.setTc(in.readFloat());else if(t==5&&f==9)x.setSoh(in.readFloat());else if(t==5&&f==10)x.setSoc(in.readFloat());else skip(in,t);});return x; }
|
/** 解析电表实际字段。 */
|
private UdanEmsDeviceState.ElectricMeterState parseMeter(byte[] b) throws IOException { UdanEmsDeviceState.ElectricMeterState x = new UdanEmsDeviceState.ElectricMeterState(); read(b,(f,t,in)->{if(t==2&&f==2)x.setDeviceId(in.readString());else if(t==1&&f==17)x.setCumulativeImportKwh(in.readDouble());else if(t==1&&f==22)x.setCumulativeExportKwh(in.readDouble());else if(t==5&&f==54)x.setAcTotalActivePower(in.readFloat());else skip(in,t);});return x; }
|
/** 解析光伏实际字段。 */
|
private UdanEmsDeviceState.PvState parsePv(byte[] b) throws IOException { UdanEmsDeviceState.PvState x = new UdanEmsDeviceState.PvState(); read(b,(f,t,in)->{if(t==2&&f==1)x.setDeviceId(in.readString());else if(t==5&&f==18)x.setActivePower(in.readFloat());else if(t==5&&f==19)x.setReactivePower(in.readFloat());else if(t==5&&f==20)x.setPowerFactor(in.readFloat());else if(t==5&&f==21)x.setDailyPowerGeneration(in.readFloat());else if(t==1&&f==22)x.setTotalPowerGeneration(in.readDouble());else skip(in,t);});return x; }
|
/** 解析充电桩实际字段。 */
|
private UdanEmsDeviceState.ChargeStationState parseChargeStation(byte[] b) throws IOException { UdanEmsDeviceState.ChargeStationState x = new UdanEmsDeviceState.ChargeStationState(); read(b,(f,t,in)->{if(t==2&&f==1)x.setDeviceId(in.readString());else if(t==0&&f==2)x.setTime(in.readUInt64());else skip(in,t);});return x; }
|
|
/** 遍历嵌套 protobuf 字段，并把字段号、线类型和读取器交给解析器。 */
|
private void read(byte[] b, FieldReader reader) throws IOException { CodedInputStream in=CodedInputStream.newInstance(b); while(!in.isAtEnd()){int tag=in.readTag();if(tag==0)break;reader.read(WireFormat.getTagFieldNumber(tag),WireFormat.getTagWireType(tag),in);} }
|
/** 跳过当前不需要映射的字段。 */
|
private void skip(CodedInputStream in, int type) throws IOException {
|
switch (type) {
|
case WireFormat.WIRETYPE_VARINT: in.readInt64(); break;
|
case WireFormat.WIRETYPE_FIXED64: in.readFixed64(); break;
|
case WireFormat.WIRETYPE_LENGTH_DELIMITED: in.readBytes(); break;
|
case WireFormat.WIRETYPE_FIXED32: in.readFixed32(); break;
|
default: break;
|
}
|
}
|
@FunctionalInterface private interface FieldReader { void read(int field, int type, CodedInputStream input) throws IOException; }
|
}
|
