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.emind.domain.EmindStation;
import com.ruoyi.other.emind.dto.EmindIncomeHistoryDto;
import com.ruoyi.other.emind.mapper.EmindStationMapper;
import com.ruoyi.other.emind.mapper.EmindStorageDashboardMapper;
import com.ruoyi.other.udan.model.EmsState;
import com.ruoyi.other.udan.model.PowerMeterState;
import com.ruoyi.other.udan.model.UdanEmsDeviceState;
import com.ruoyi.other.udan.util.UdanEmsKafkaUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;

/** 将 IEmsAllDeviceState 中模型已有的 EMS 和能耗计字段转换为业务对象。 */
@Slf4j
@Component
@ConditionalOnProperty(prefix = "udan.ems.kafka", name = "enabled", havingValue = "true")
public class UdanEmsMessageProcessor {
    private final UdanEmsKafkaUtil kafkaUtil;
    private final EmindStorageDashboardMapper dashboardMapper;
    private final EmindStationMapper stationMapper;

    public UdanEmsMessageProcessor(UdanEmsKafkaUtil kafkaUtil, EmindStorageDashboardMapper dashboardMapper, EmindStationMapper stationMapper) {
        this.kafkaUtil = kafkaUtil;
        this.dashboardMapper = dashboardMapper;
        this.stationMapper = stationMapper;
    }

    @PostConstruct
    public void register() {
        kafkaUtil.setMessageHandler(this::process);
    }

    public void process(String id, byte[] bytes, String emsDeviceId) {
        try {
            LocalDateTime now = LocalDateTime.now();
            UdanEmsDeviceState udanEmsDeviceState = parse(id, emsDeviceId, bytes);
            if(StringUtils.isNotEmpty(udanEmsDeviceState.getEmsDeviceId())) {
                log.info("2、UDAN EMS实际对象: {}", JSON.toJSONString(udanEmsDeviceState));
                List<EmindStation>  stations = stationMapper.selectRemoteActiveStations();
                EmindStation station = stations.stream().filter(s -> udanEmsDeviceState.getEmsDeviceId().equals(s.getStationCode())).findFirst().orElse(null);
                if(Objects.nonNull(station)) {
                    List<EmindIncomeHistoryDto> daily = udanEmsDeviceState.convert(station.getStationId());
                    Set<LocalDate> dateSet = daily.stream().map(EmindIncomeHistoryDto::getDate).collect(Collectors.toSet());
                    // 删除电站下该天已存在的旧数据
                    if(CollectionUtils.isNotEmpty(dateSet)) {
                        dashboardMapper.deleteByStationIdDay(station.getStationId(), dateSet);
                    }
                    if(CollectionUtils.isNotEmpty(daily)) {
                        dashboardMapper.upsertDaily(station.getStationId(), daily, now);
                    }
                }else{
                    log.error("UDAN EMS数据同步错误-电站信息不存在！");
                }
            }

        } catch (IOException e) {
            log.error("UDAN EMS protobuf解析失败: emsSystemId={}, bytes={}", id,
                    bytes == null ? 0 : bytes.length, e);
        }
    }

    /** 兼容不带 Header 的直接调用。 */
    public void process(String id, byte[] bytes) {
        process(id, bytes, null);
    }

    public UdanEmsDeviceState parse(String id, String emsDeviceId, byte[] bytes) throws IOException {
        UdanEmsDeviceState state = new UdanEmsDeviceState();
        state.setEmsSystemId(id);
        state.setEmsDeviceId(emsDeviceId);
        read(bytes, (field, type, in) -> {
            if (field == 1 && type == 0) state.setTime(in.readUInt64());
            else if (type == 2 && field == 2) state.setEmsState(parseEms(in.readBytes().toByteArray()));
            else if (type == 2 && field == 8) state.setPowerMeterState(parsePowerMeter(in.readBytes().toByteArray()));
            else skip(in, type);
        });
        return state;
    }

    /** 兼容不带 Header 的直接解析调用。 */
    public UdanEmsDeviceState parse(String id, byte[] bytes) throws IOException {
        return parse(id, null, bytes);
    }

    private EmsState parseEms(byte[] bytes) throws IOException {
        EmsState x = new EmsState();
        read(bytes, (f, t, in) -> {
            switch (f) {
                case 1: x.setDeviceId(str(in,t)); break; case 9: x.setEmsStatus(enumName(in,t,"RunStatus")); break;
                case 18: x.setDemandPower(f32(in,t)); break; case 20: x.setTime(u64(in,t)); break; case 21: x.setProtoProtocolVersion(str(in,t)); break;
                case 22: x.setEmsRunStatus(enumAt(EmsState.EmsRunStatus.values(),in,t)); break; case 23: x.setEmsWorkStatus(enumAt(EmsState.EmsWorkStatus.values(),in,t)); break;
                case 24: x.setFaultStatus(enumAt(EmsState.ErrorStatus.values(),in,t)); break; case 25: x.setOnGridStatus(enumName(in,t,"GridStatus")); break;
                case 26: x.setControlSrc(enumAt(EmsState.ControlSrc.values(),in,t)); break; case 27: x.setEmuModeCode(enumAt(EmsState.EmuModeCode.values(),in,t)); break;
                case 28: x.setEmuAutoMod(enumAt(EmsState.EmuAutoMode.values(),in,t)); break; case 29: x.setOverDischargeProtect(enumAt(EmsState.YesNo.values(),in,t)); break;
                case 30: x.setOverChargeProtect(enumAt(EmsState.YesNo.values(),in,t)); break; case 31: x.setDemandControl(enumAt(EmsState.YesNo.values(),in,t)); break; case 32: x.setCapacityExpansionModeCtrl(enumAt(EmsState.YesNo.values(),in,t)); break;
                case 33: x.setDailyChargeEnergyKwh(f32(in,t)); break; case 34: x.setDailyDischargeEnergyKwh(f32(in,t)); break; case 35: x.setMonthlyChargeEnergyKwh(f32(in,t)); break; case 36: x.setMonthlyDischargeEnergyKwh(f32(in,t)); break;
                case 37: x.setTotalChargeEnergyKwh(f64(in,t)); break; case 38: x.setTotalDischargeEnergyKwh(f64(in,t)); break; case 39: x.setPcsSumsActivePower(f32(in,t)); break; case 40: x.setPcsSumsApparentPower(f32(in,t)); break;
                case 41: x.setDcBusVoltage(f32(in,t)); break; case 42: x.setDcBusCurrent(f32(in,t)); break; case 43: x.setSumsSoc(f32(in,t)); break; case 44: x.setCellSums(u32(in,t)); break;
                case 45: x.setBatteryAverageTempr(f32(in,t)); break; case 46: x.setBatteryMaxTemprDiff(f32(in,t)); break; case 47: x.setMaxCellVoltage(u32(in,t)); break; case 48: x.setMinCellVoltage(u32(in,t)); break;
                case 49: x.setSystemMaxAllowChgPower(f32(in,t)); break; case 50: x.setSystemMaxAllowDisPower(f32(in,t)); break; case 51: x.setRechargeableKwh(f32(in,t)); break; case 52: x.setDischargeableKwh(f32(in,t)); break;
                case 53: x.setCurrentMaxAllowChgPower(f32(in,t)); break; case 54: x.setCurrentMaxAllowDisPower(f32(in,t)); break; case 55: x.setSocUpperLimit(f32(in,t)); break; case 56: x.setSocLowerLimit(f32(in,t)); break;
                case 57: x.setMaxPowerDischargeDuration(u32(in,t)); break; case 58: x.setMaxPowerChargeDuration(u32(in,t)); break; case 60: x.setMasterCommStatus(enumName(in,t,"CommStatus")); break;
                case 61: x.setGridImportActiveEnergyKwh(f64(in,t)); break; case 62: x.setGridExportActiveEnergyKwh(f64(in,t)); break; case 63: x.setDcTotalPower(f32(in,t)); break; case 64: x.setLoadTotalPower(f32(in,t)); break;
                case 65: x.setEnergyStoreTotalPower(f32(in,t)); break; case 66: x.setHighSideGridPower(f32(in,t)); break; case 67: x.setLowSideGridPower(f32(in,t)); break; case 68: x.setPvPower(f32(in,t)); break; case 69: x.setSumsSoh(f32(in,t)); break;
                default: skip(in,t);
            }
        });
        return x;
    }

    private PowerMeterState parsePowerMeter(byte[] bytes) throws IOException {
        PowerMeterState value = new PowerMeterState();
        read(bytes, (field, type, input) -> {
            switch (field) {
                case 1: value.setCumulativeImportAmountPeak(f64(input, type)); break;
                case 2: value.setCumulativeImportAmountShoulder(f64(input, type)); break;
                case 3: value.setCumulativeImportAmountValley(f64(input, type)); break;
                case 4: value.setCumulativeImportAmount(f64(input, type)); break;
                case 5: value.setDailyImportAmount((float) f64(input, type)); break;
                case 6: value.setMonthlyImportAmount(f64(input, type)); break;
                case 7: value.setDailyProfit(f64(input, type)); break;
                case 8: value.setYesterdayProfit(f64(input, type)); break;
                case 9: value.setMonthlyProfit(f64(input, type)); break;
                case 10: value.setLastMonthlyProfit(f64(input, type)); break;
                case 11: value.setCumulativeProfit(f64(input, type)); break;
                case 12: value.setCumulativeImportAmountSharp(f64(input, type)); break;
                case 15: value.setCumulativeImportKwh(f64(input, type)); break;
                case 16: value.setCumulativeImportKwhSharp(f64(input, type)); break;
                case 17: value.setCumulativeImportKwhPeak(f64(input, type)); break;
                case 18: value.setCumulativeImportKwhShoulder(f64(input, type)); break;
                case 19: value.setCumulativeImportKwhValley(f64(input, type)); break;
                case 20: value.setCumulativeExportKwh(f64(input, type)); break;
                case 21: value.setCumulativeExportKwhSharp(f64(input, type)); break;
                case 22: value.setCumulativeExportKwhPeak(f64(input, type)); break;
                case 23: value.setCumulativeExportKwhShoulder(f64(input, type)); break;
                case 24: value.setCumulativeExportKwhValley(f64(input, type)); break;
                case 25: value.setMonthlyImportKwh(f64(input, type)); break;
                case 26: value.setMonthlyImportKwhSharp(f64(input, type)); break;
                case 27: value.setMonthlyImportKwhPeak(f64(input, type)); break;
                case 28: value.setMonthlyImportKwhShoulder(f64(input, type)); break;
                case 29: value.setMonthlyImportKwhValley(f64(input, type)); break;
                case 30: value.setMonthlyExportKwh(f64(input, type)); break;
                case 31: value.setMonthlyExportKwhSharp(f64(input, type)); break;
                case 32: value.setMonthlyExportKwhPeak(f64(input, type)); break;
                case 33: value.setMonthlyExportKwhShoulder(f64(input, type)); break;
                case 34: value.setMonthlyExportKwhValley(f64(input, type)); break;
                case 35: value.setDailyImportKwh(f32(input, type)); break;
                case 36: value.setDailyImportKwhSharp(f32(input, type)); break;
                case 37: value.setDailyImportKwhPeak(f32(input, type)); break;
                case 38: value.setDailyImportKwhShoulder(f32(input, type)); break;
                case 39: value.setDailyImportKwhValley(f32(input, type)); break;
                case 40: value.setDailyExportKwh(f32(input, type)); break;
                case 41: value.setDailyExportKwhSharp(f32(input, type)); break;
                case 42: value.setDailyExportKwhPeak(f32(input, type)); break;
                case 43: value.setDailyExportKwhShoulder(f32(input, type)); break;
                case 44: value.setDailyExportKwhValley(f32(input, type)); break;
                case 45: value.setCumulativeExportAmount(f64(input, type)); break;
                case 46: value.setCumulativeExportAmountSharp(f64(input, type)); break;
                case 47: value.setCumulativeExportAmountPeak(f64(input, type)); break;
                case 48: value.setCumulativeExportAmountShoulder(f64(input, type)); break;
                case 49: value.setCumulativeExportAmountValley(f64(input, type)); break;
                case 50: value.setMonthlyImportAmountSharp(f64(input, type)); break;
                case 51: value.setMonthlyImportAmountPeak(f64(input, type)); break;
                case 52: value.setMonthlyImportAmountShoulder(f64(input, type)); break;
                case 53: value.setMonthlyImportAmountValley(f64(input, type)); break;
                case 54: value.setMonthlyExportAmount(f64(input, type)); break;
                case 55: value.setMonthlyExportAmountSharp(f64(input, type)); break;
                case 56: value.setMonthlyExportAmountPeak(f64(input, type)); break;
                case 57: value.setMonthlyExportAmountShoulder(f64(input, type)); break;
                case 58: value.setMonthlyExportAmountValley(f64(input, type)); break;
                case 59: value.setDailyImportAmountSharp(f32(input, type)); break;
                case 60: value.setDailyImportAmountPeak(f32(input, type)); break;
                case 61: value.setDailyImportAmountShoulder(f32(input, type)); break;
                case 62: value.setDailyImportAmountValley(f32(input, type)); break;
                case 63: value.setDailyExportAmount(f32(input, type)); break;
                case 64: value.setDailyExportAmountSharp(f32(input, type)); break;
                case 65: value.setDailyExportAmountPeak(f32(input, type)); break;
                case 66: value.setDailyExportAmountShoulder(f32(input, type)); break;
                case 67: value.setDailyExportAmountValley(f32(input, type)); break;
                case 68: value.setTime(u64(input, type)); break;
                default: skip(input, type);
            }
        });
        return value;
    }

    private void read(byte[] bytes, Reader reader) throws IOException {
        CodedInputStream input = CodedInputStream.newInstance(bytes == null ? new byte[0] : bytes);
        while (!input.isAtEnd()) {
            int tag = input.readTag();
            if (tag == 0) {
                break;
            }
            reader.read(WireFormat.getTagFieldNumber(tag), WireFormat.getTagWireType(tag), input);
        }
    }

    private static String str(CodedInputStream input, int type) throws IOException {
        if (type == WireFormat.WIRETYPE_LENGTH_DELIMITED) {
            return input.readString();
        }
        skip(input, type);
        return null;
    }

    private static long u64(CodedInputStream input, int type) throws IOException {
        if (type == WireFormat.WIRETYPE_VARINT) {
            return input.readUInt64();
        }
        skip(input, type);
        return 0L;
    }

    private static int u32(CodedInputStream input, int type) throws IOException {
        return (int) u64(input, type);
    }

    private static float f32(CodedInputStream input, int type) throws IOException {
        if (type == WireFormat.WIRETYPE_FIXED32) {
            return input.readFloat();
        }
        skip(input, type);
        return 0F;
    }

    private static double f64(CodedInputStream input, int type) throws IOException {
        if (type == WireFormat.WIRETYPE_FIXED64) {
            return input.readDouble();
        }
        skip(input, type);
        return 0D;
    }
    private static String enumName(CodedInputStream input, int type, String prefix) throws IOException {
        if (type != WireFormat.WIRETYPE_VARINT) {
            skip(input, type);
            return null;
        }
        int number = input.readInt32();
        if ("RunStatus".equals(prefix)) {
            return new String[]{"RUN_STATUS_UNKNOWN", "RUN_STATUS_STOPPED", "RUN_STATUS_RUNNING"}[number < 0 || number > 2 ? 0 : number];
        }
        if ("GridStatus".equals(prefix)) {
            return new String[]{"GRID_STATUS_UNKNOWN", "GRID_STATUS_ON_GRID", "GRID_STATUS_OFF_GRID"}[number < 0 || number > 2 ? 0 : number];
        }
        if ("CommStatus".equals(prefix)) {
            return new String[]{"COMM_STATUS_UNKNOWN", "COMM_STATUS_DISCONNECTED", "COMM_STATUS_CONNECTED"}[number < 0 || number > 2 ? 0 : number];
        }
        return prefix + "_" + number;
    }
    private static <T> T enumAt(T[] values, CodedInputStream input, int type) throws IOException {
        int number = (int) u64(input, type);
        return number >= 0 && number < values.length ? values[number] : null;
    }

    private static void skip(CodedInputStream input, int type) throws IOException {
        switch (type) {
            case WireFormat.WIRETYPE_VARINT:
                input.readInt64();
                break;
            case WireFormat.WIRETYPE_FIXED64:
                input.readFixed64();
                break;
            case WireFormat.WIRETYPE_LENGTH_DELIMITED:
                input.readBytes();
                break;
            case WireFormat.WIRETYPE_FIXED32:
                input.readFixed32();
                break;
            default:
                throw new IOException("Unsupported protobuf wire type: " + type);
        }
    }

    @FunctionalInterface
    private interface Reader {
        void read(int field, int type, CodedInputStream input) throws IOException;
    }
}