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; }
}