package com.yeejoin.equipmanage.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.yeejoin.amos.boot.biz.common.utils.RedisKey;
import com.yeejoin.amos.boot.biz.common.utils.RedisUtils;
import com.yeejoin.amos.component.influxdb.InfluxDbConnection;
import com.yeejoin.amos.component.rule.RuleTrigger;
import com.yeejoin.amos.feign.systemctl.model.MessageModel;
import com.yeejoin.equipmanage.common.datasync.entity.FireEquipmentDefectAlarm;
import com.yeejoin.equipmanage.common.datasync.entity.FireEquipmentFaultAlarm;
import com.yeejoin.equipmanage.common.datasync.entity.FireEquipmentFireAlarm;
import com.yeejoin.equipmanage.common.dto.TemperatureAlarmDto;
import com.yeejoin.equipmanage.common.dto.EquipIndexDto;
import com.yeejoin.equipmanage.common.entity.*;
import com.yeejoin.equipmanage.common.entity.dto.EquipQrDateDto;
import com.yeejoin.equipmanage.common.entity.publics.BaseEntity;
import com.yeejoin.equipmanage.common.entity.vo.*;
import com.yeejoin.equipmanage.common.enums.*;
import com.yeejoin.equipmanage.common.utils.DateUtils;
import com.yeejoin.equipmanage.common.utils.StringUtil;
import com.yeejoin.equipmanage.common.utils.UUIDUtils;
import com.yeejoin.equipmanage.common.vo.*;
import com.yeejoin.equipmanage.fegin.IotFeign;
import com.yeejoin.equipmanage.fegin.SystemctlFeign;
import com.yeejoin.equipmanage.mapper.*;
import com.yeejoin.equipmanage.quartz.PumpSendMessage;
import com.yeejoin.equipmanage.quartz.QuartzManager;
import com.yeejoin.equipmanage.remote.RemoteSecurityService;
import com.yeejoin.equipmanage.service.*;
import com.yeejoin.equipmanage.utils.BeanUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.IOUtils;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.scheduling.annotation.Async;
import org.springframework.core.io.Resource;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import org.typroject.tyboot.component.emq.EmqKeeper;
import org.typroject.tyboot.core.foundation.utils.ValidationUtil;
import org.typroject.tyboot.core.restful.exception.instance.BadRequest;
import org.typroject.tyboot.core.restful.utils.ResponseModel;

import java.io.IOException;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.math.RoundingMode;
import java.nio.charset.StandardCharsets;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.stream.Collectors;

import static com.yeejoin.equipmanage.common.enums.MqttConstant.*;


/**
 * @author keyong
 * @title: MqttReceiveServiceImpl
 *
 * <pre>
 * &#64;description: 增量数据处理
 *         </pre>
 * @date 2020/11/3 13:39
 */
@Slf4j
@Service
public class MqttReceiveServiceImpl implements MqttReceiveService {
    /**
     * 泡沫罐KEY
     */
    private final static String CAFS_FoamTank_FoamTankLevel = "CAFS_FoamTank_FoamTankLevel";
    /**
     * 泡沫罐KEY
     */
    private final static String FHS_PipePressureDetector_PipePressure = "FHS_PipePressureDetector_PipePressure";
    /**
     * 水池信息
     */
    private final static String FHS_FirePoolDevice_WaterLevel = "FHS_FirePoolDevice_WaterLevel";
    /**
     * 水池信息
     */
    private final static String FHS_WirelessliquidDetector_WaterLevel = "FHS_WirelessliquidDetector_WaterLevel";
    /**
     * 水箱液位
     */
    private final static String CAFS_WaterTank_WaterTankLevel = "CAFS_WaterTank_WaterTankLevel";
    private static final String PUMP_JOB_GROUP_NAME = "EQUIP_PUMP_JOB_GROUP_NAME";
    private static final String PUMP_TRIGGER_NAME = "EQUIP_PUMP_TRIGGER_NAME";
    private static final String PUMP_TRIGGER_GROUP_NAME = "EQUIP_PUMP_TRIGGER_GROUP_NAME";
    private static final Map<String, TemperatureAlarmDto> temperatureMap = new HashMap<>();
    static IEquipmentSpecificIndexService equipmentSpecificIndexService;
    static EquipmentSpecificMapper equipmentSpecificMapper;
    static IFireFightingSystemService fireFightingSystemService;
    static MqttSendGateway mqttSendGateway;
    private static RemoteSecurityService remoteSecurityService;
    private static String canvasTopic;
    private static Boolean jcsSwitch;
    private static Boolean bool = Boolean.FALSE;
    @Autowired
    protected EmqKeeper emqKeeper;
    @Autowired
    ICarPropertyService carPropertyService;
    @Autowired
    IEquipmentSpecificAlarmService equipmentSpecificAlarmService;
    @Autowired
    IEquipmentSpecificAlarmLogService equipmentSpecificAlarmLogService;
    @Autowired
    EquipmentSpecificAlarmMapper equipmentSpecificAlarmMapper;
    @Autowired
    EquipQrcodeRecordMapper equipQrcodeRecordMapper;

    @Autowired
    EquipmentSpecificAlarmLogMapper equipmentSpecificAlarmLogMapper;
    @Autowired
    @Lazy
    IEquipmentIndexService equipmentIndexService;
    @Autowired
    EquipmentSpecificIndexMapper equipmentSpecificIndexMapper;
    @Autowired
    MarqueeDataMapper marqueeDataMapper;
    //消防泵
    @Value("${equipment.plan.pump}")
    String pumpCodes;
    //消防炮
    @Value("${equipment.plan.monitor}")
    String monitorCodes;
    @Autowired
    private InfluxDbConnection influxDbConnection;

    @Autowired
    private RuleTrigger ruleTrigger;

    @Autowired
    CarMapper carMapper;
    @Autowired
    FireFightingSystemMapper fireFightingSystemMapper;
    @Autowired
    IESeqService eSeqService;
    @Autowired
    private IotFeign iotFeign;
    @Autowired
    private RedisUtils redisUtils;
    @Autowired
    private ISyncDataService syncDataService;
    @Autowired
    private IEquipmentAlarmReportDayService iEquipmentAlarmReportDayService;
    @Autowired
    private IEquipmentSpecificSerivce iEquipmentSpecificSerivce;
    @Autowired
    private SystemctlFeign systemctlFeign;
    @Autowired
    private TopographyService topographyService;
    @Autowired
    private IEquipmentService equipmentService;
    @Autowired
    private IPressurePumpService pressurePumpService;

    @Autowired
    EquipmentDetailMapper equipmentDetailMapper;

    @Autowired
    private ManufacturerInfoMapper manufacturerInfoMapper;
//    @Autowired
//    private AmosRequestContext amosAuth;
    @Value("${equipManage.name}")
    private String serverName;
    @Value("${state.code:code}")
    private String stationCode;
    @Value("${system.type}")
    private String system;
    @Value("${state.name:name}")
    private String stationName;
    @Value("${mqtt.vehicle.topic}")
    private String carTopic;

    @Value("${equip.risk.model.topic}")
    private String riskTopic;

    @Value("${equip.message.create.topic}")
    private String riskTopicCreate;

    @Value("${equip.center.risk.topic}")
    private String riskTopicCenter;

    @Value("${equip.index.topic}")
    private String indexTopic;
    @Value("${spring.redis.expire.time}")
    private long redisExpireTime;
    @Value("${systemctl.sync.switch}")
    private Boolean syncSwitch;
    @Value("${dcs.send.message}")
    private Boolean NRMsgSend;
    @Value("${systemctl.amos.switch}")
    private Boolean amosSwitch;
    @Value("${isSendApp}")
    private Boolean isSendApp;
    @Value("${isSendIot:false}")
    private Boolean isSendIot;
    @Value("${is.open.telemetering:false}")
    private Boolean isOpenTelemetering;
    @Value("${equipment.pressurepump.start}")
    private String pressurePumpStart;
    @Value("${emergency.disposal.indicators}")
    private String emergencyDisposalIndicators;
    @Value("classpath:/json/systemStateIndex.json")
    private Resource systemStateIndex;
    @Value("${message.alarm.char}")
    private String specialChars;

    public static List<EquipmentSpecificAlarmLog> upAlarmLogStatus(String iotCode, String equipmentSpecificIndexKey,
                                                                   String traceId, IEquipmentSpecificAlarmLogService equipmentSpecificAlarmLogService,
                                                                   boolean flag, boolean batchConfirm, String confirmType) {

//        LambdaQueryWrapper<EquipmentSpecificAlarmLog> queryWrapper = new LambdaQueryWrapper<>();
//        if (batchConfirm.equalsIgnoreCase(HandleBatchConfirmEnum.SINGLE.getCode())) {
//            queryWrapper.eq(EquipmentSpecificAlarmLog::getIotCode, iotCode);
//            queryWrapper.ne(EquipmentSpecificAlarmLog::getStatus, AlarmStatusEnum.HF.getCode());
//        } else if (batchConfirm.equalsIgnoreCase(HandleBatchConfirmEnum.ALL.getCode())) {
//            // 此处做了全库的设备告警消除原因有二：第一点 目前在警情确认的时候会把设备定义类型是确认消除的设备的同一类型告警全部消除，所以此处逻辑合理；
//            //    第二点：现在做了一键做全库设备的确认操作，但是没有全库设备的消除操作功能，所以此处做如下逻辑符合功能操作的惯性。     ----2023年9月20日 柯勇
//            queryWrapper.ne(EquipmentSpecificAlarmLog::getStatus, AlarmStatusEnum.HF.getCode());
//        } else {
//            queryWrapper.eq(EquipmentSpecificAlarmLog::getIotCode, iotCode);
//            queryWrapper.eq(EquipmentSpecificAlarmLog::getEquipmentSpecificIndexKey, equipmentSpecificIndexKey);
//            queryWrapper.ne(EquipmentSpecificAlarmLog::getStatus, AlarmStatusEnum.HF.getCode());
//        }
        LambdaQueryWrapper<EquipmentSpecificAlarmLog> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(EquipmentSpecificAlarmLog::getIotCode, iotCode);
        queryWrapper.eq(EquipmentSpecificAlarmLog::getEquipmentSpecificIndexKey, equipmentSpecificIndexKey);
        queryWrapper.ne(EquipmentSpecificAlarmLog::getStatus, AlarmStatusEnum.HF.getCode());
        List<EquipmentSpecificAlarmLog> logs = equipmentSpecificAlarmLogService.getBaseMapper().selectList(queryWrapper);
        if (!logs.isEmpty()) {
            EquipmentSpecificAlarmLog log = logs.get(0);
            EquipmentSpecific specific = equipmentSpecificMapper.selectById(log.getEquipmentSpecificId());
            Date date = new Date();
            SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            String dateString = formatter.format(date);
            logs.forEach(x -> {
                x.setCleanTime(new Date());
                x.setStatus(AlarmStatusEnum.HF.getCode());
                if (!flag) {
                    // 若已经被确警处理过，此处不再做处理
                    if (StringUtils.isEmpty(x.getConfirmType())) {
                        x.setConfirmType(x.getType());
                        x.setConfirmTypeName(ConfirmAlamEnum.getTypeByCode(x.getType()));
                        String equipmentName = StringUtil.isNotEmpty(specific.getCode()) ? specific.getName() + "(" + specific.getCode() + ")" : specific.getName();
                        if (x.getType().equals(AlarmTypeEnum.PB.getCode())) {
                            x.setAlarmReason(equipmentName + "频繁故障/误报，将设备报警信息屏蔽。");
                        } else {
                            x.setAlarmReason(equipmentName + "引起设备报警");
                        }
                        x.setResolveResult(dateString + "系统收到设备复归（已消除）信号，系统自动处理。");
                        x.setConfirmUser("");
                        x.setConfirmUserName("系统自动处理");
                        x.setConfirmDate(date);
                    }
                }
                x.setEquipmentSpecificIndexValue(TrueOrFalseEnum.fake.value);
                if (!StringUtils.isEmpty(traceId)) {
                    x.setTraceId(traceId);
                }
            });
            equipmentSpecificAlarmLogService.updateBatchById(logs);
        }
        if (flag) {
            List<EquipmentSpecificIndex> indexList = equipmentSpecificIndexService.getEquipmentSpeIndexBySpeIotCode(iotCode);
            publishDataToCanvas(indexList);
        }
        return logs;
    }

    /**
     * 画布数据消息推送
     *
     * @param indexList
     */
    public static void publishDataToCanvas(List<EquipmentSpecificIndex> indexList) {
        if (!ObjectUtils.isEmpty(indexList)) {
            EquipmentSpecificIndex index = indexList.stream().filter(x -> x.getUpdateDate() != null)
                    .sorted(Comparator.comparing(EquipmentSpecificIndex::getUpdateDate).reversed())
                    .collect(Collectors.toList()).get(0);
            EquipmentStateVo equipmentStateVo = new EquipmentStateVo();
            equipmentStateVo.setEquipName(index.getEquipmentSpecificName());
            equipmentStateVo.setOrgCode(index.getOrgCode());
            equipmentStateVo.setSpecificId(index.getEquipmentSpecificId());
            equipmentStateVo.setEquipCode(index.getQrCode());
            equipmentStateVo.setEquipIotCode(index.getIotCode());
            equipmentStateVo.setStatus("");
            equipmentStateVo.setColor(index.getEmergencyLevelColor());
            // 添加性能指标项
            //flag 无意义 getSpeIndex 方法其他地方共用 做了额外封装 为0时走原逻辑
            int flag = 0;
            equipmentStateVo.setSpeindexList(fireFightingSystemService.getSpeIndex(index.getEquipmentSpecificId(), flag));
            Map<String, Object> topicObject = new HashMap<>();
            topicObject.put("equipCode", equipmentStateVo.getEquipCode());
            if (TrueOrFalseEnum.real.value.equalsIgnoreCase(index.getValue())) {
                topicObject.put("color", equipmentStateVo.getColor());
            } else {
                topicObject.put("color", "");
            }
            Map<String, Object> map = JSON.parseObject(JSON.toJSONString(equipmentStateVo));
            topicObject.put("uuid", UUID.randomUUID().toString());
            map.put(canvasTopic, topicObject);

            // 发送数据至画布
            mqttSendGateway.sendToMqtt(canvasTopic, JSON.toJSONString(map));
        }
    }

    @Autowired
    public void setEquipmentSpecificIndexService(IEquipmentSpecificIndexService equipmentSpecificIndexService) {
        MqttReceiveServiceImpl.equipmentSpecificIndexService = equipmentSpecificIndexService;
    }

    @Autowired
    public void setEquipmentSpecificMapper(EquipmentSpecificMapper equipmentSpecificMapper) {
        MqttReceiveServiceImpl.equipmentSpecificMapper = equipmentSpecificMapper;
    }

    @Value("${equip.message.create.topic}")
    private String riskMsgCreateTopic;

    @Autowired
    public void setFireFightingSystemService(IFireFightingSystemService fireFightingSystemService) {
        MqttReceiveServiceImpl.fireFightingSystemService = fireFightingSystemService;
    }

    private static Boolean isAlarm = Boolean.FALSE;

    @Autowired
    public void setMqttSendGateway(MqttSendGateway mqttSendGateway) {
        MqttReceiveServiceImpl.mqttSendGateway = mqttSendGateway;
    }

    @Autowired
    public void setRemoteSecurityService(RemoteSecurityService remoteSecurityService) {
        MqttReceiveServiceImpl.remoteSecurityService = remoteSecurityService;
    }

    @Value("${equip.point.equipmentdata.topic}")
    public void setCanvasTopic(String canvasTopic) {
        MqttReceiveServiceImpl.canvasTopic = canvasTopic;
    }

    @Value("${systemctl.jcs.switch}")
    public void setJcsSwitch(Boolean jcsSwitch) {
        MqttReceiveServiceImpl.jcsSwitch = jcsSwitch;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void handlerMqttIncrementMessage(String topic, String message) {

        TopicEntityVo topicEntity = new TopicEntityVo();
        topicEntity.setTopic(topic);
        topicEntity.setMessage(message);
        int endIndex = topic.lastIndexOf("/");
        String iotCode = topic.substring(0, endIndex).replace("/", "");
        topicEntity.setIotCode(iotCode);
        List<EquipmentSpecificVo> eqIotCodeList = iEquipmentSpecificSerivce.getEquipAndCarIotcodeByIotcode(iotCode);
        if (eqIotCodeList.isEmpty()) {
            log.info("该数据{}不存在！", iotCode);
            return;
        }
        if (eqIotCodeList.size() > 1) {
            log.info("有重复的{}数据！", iotCode);
        }
        //给 iot服务 推送消息 插数据到 influxdb
        if (isSendIot) {
            mqttSendGateway.sendToMqtt("influxdb/" + topic.substring(0, endIndex), message);
        }
        EquipmentSpecificVo vo = eqIotCodeList.get(0);
        topicEntity.setType(vo.getType());
        topicEntity.setCode(vo.getCode());

        JSONObject json = JSONObject.parseObject(message);
        Iterator it = json.entrySet().iterator();
        List<IotDataVO> iotDatalist = new ArrayList<>();
        String traceId = "";
        while (it.hasNext()) {
            IotDataVO iotDataVO = new IotDataVO();
            Map.Entry<String, Object> entry = (Map.Entry<String, Object>) it.next();
            String key = entry.getKey();
            Object value = entry.getValue();
            iotDataVO.setKey(key);
            iotDataVO.setValue(value);
            if ("traceId".equalsIgnoreCase(key)) {
                traceId = value.toString();
                continue;
            }
            iotDatalist.add(iotDataVO);
        }
        if (ObjectUtils.isEmpty(iotDatalist)) {
            return;
        }
        log.info(String.format("收到mqtt消息：%s", message));

        // 发送emq消息转kafka
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("topic", topic);
        jsonObject.put("data", message);

        try {
            emqKeeper.getMqttClient().publish("emq.iot.created", jsonObject.toString().getBytes(), 1, false);
        } catch (MqttException e) {
            log.info(String.format("发送eqm转kafka消息失败：%s", e.getMessage()));
        }

        if (!StringUtils.isEmpty(traceId)) {
            String finalTraceId = traceId;
            List<IotDataVO> collect = iotDatalist.stream().map(x -> {
                x.setTraceId(finalTraceId);
                return x;
            }).collect(Collectors.toList());
            realTimeDateProcessing(topicEntity, collect, vo);
        } else {
            realTimeDateProcessing(topicEntity, iotDatalist, vo);
        }
    }

    @Override
    @Async("equipAsyncExecutor")
    @Transactional(rollbackFor = Exception.class)
    public void handlerMqttRomaMessage(String topic, String message) {
        TopicEntityVo topicEntity = new TopicEntityVo();
        topicEntity.setTopic(topic);
        topicEntity.setMessage(message);

        List<IotDataVO> iotDatalist = new ArrayList<>();
        List<EquipmentSpecificIndex> equipmentSpecificIndexList = new ArrayList<>();
        List<EquipmentSpecificAlarm> equipmentSpecificAlarms = new ArrayList<>();
        List<IndexStateVo> indexStateList = new ArrayList<>();
        JSONObject messageObj = JSONObject.parseObject(message);
        JSONObject data = messageObj.getJSONObject("data");
        if (!ObjectUtils.isEmpty(data)) {
            message = JSON.toJSONString(data);
        }

        StationMessage stationMessage = JSON.parseObject(message, StationMessage.class);
        log.info("接收到换流站Kafka消息: {}", stationMessage);
        String indexAddress = null, value = null, timeStamp = null, quality = null, dataType = null;
        if (stationMessage != null) {
            dataType = stationMessage.getDataType();
            timeStamp = stationMessage.getTimeStamp();
            quality = stationMessage.getQuality();

            if (dataType.equals(STATE)) {
                indexAddress = stationMessage.getScadaId();
                value = ONE_1.equalsIgnoreCase(stationMessage.getValue()) ? TRUE : FALSE;
            } else if (dataType.equals(DIS_CREATE)) {
                indexAddress = stationMessage.getKey();
                value = ONE_1_0.equalsIgnoreCase(stationMessage.getValue()) ? TRUE : FALSE;
            } else {
                indexAddress = stationMessage.getKey();
                value = stationMessage.getValue();
            }
        }

        Map<Object, Object> equipmentIndexKeyMap = redisUtils.hmget(RedisKey.EQUIP_INDEX_ADDRESS_KEY);
        if (equipmentIndexKeyMap.get(indexAddress) != null) {
            EquipmentSpecificIndex equipmentSpeIndex = equipmentSpecificIndexService.getEquipmentSpeIndexByAddress(indexAddress, null, null);
            EquipmentIndex equipmentIndex = equipmentIndexService.getById(equipmentSpeIndex.getEquipmentIndexId());

            equipmentSpeIndex.setValue(value);
            equipmentSpeIndex.setValueLabel(valueTranslate(value, equipmentSpeIndex.getValueEnum()));
            equipmentSpeIndex.setEquipmentType(topicEntity.getType());
            equipmentSpeIndex.setUpdateDate(new Date());
            equipmentSpeIndex.setQuality(quality);
            equipmentSpeIndex.setDataType(dataType);
            equipmentSpeIndex.setTimeStamp(timeStamp);
            equipmentSpeIndex.setUUID(UUIDUtils.getUUID());
            equipmentSpeIndex.setUnit(Objects.nonNull(equipmentIndex.getUnit()) ? equipmentIndex.getUnit() : equipmentIndex.getUnitName());

            IotDataVO iotDataVO = new IotDataVO();
            iotDataVO.setKey(equipmentSpeIndex.getNameKey());
            iotDataVO.setValue(value);
            iotDatalist.add(iotDataVO);

            QueryWrapper<EquipmentSpecific> queryWrapper = new QueryWrapper<>();
            queryWrapper.eq("id", equipmentSpeIndex.getEquipmentSpecificId());
            EquipmentSpecific equipmentSpecific = iEquipmentSpecificSerivce.getOne(queryWrapper);
            if (equipmentSpecific == null) {
                return;
            }
            String iotCode = equipmentSpecific.getIotCode();
            StringBuilder endIndex = new StringBuilder(iotCode).insert(8, '/');
            String iotTopic = "influxdb/" + endIndex;
            JSONObject msg = new JSONObject();
            msg.put(equipmentSpeIndex.getEquipmentIndexKey(), value);
            mqttSendGateway.sendToMqtt(iotTopic, JSON.toJSONString(msg));

            List<EquipmentSpecificVo> eqIotCodeList = iEquipmentSpecificSerivce.getEquipAndCarIotcodeByIotcode(iotCode);

            if (eqIotCodeList.isEmpty()) {
                log.info("该数据{}不存在！", iotCode);
                return;
            }
            if (eqIotCodeList.size() > 1) {
                log.info("有重复的{}数据！", iotCode);
            }
            EquipmentSpecificVo equipmentSpecificVo = eqIotCodeList.get(0);
            topicEntity.setType(equipmentSpecificVo.getType());
            topicEntity.setCode(equipmentSpecificVo.getCode());

            //es存储数据
            eSeqService.saveESEquiplistSpecificBySystemESVO(equipmentSpeIndex, String.valueOf(equipmentSpecificVo.getSystemId()), equipmentSpecificVo.getSystemName());
            //更新装备性能指标
            log.info("接收到遥信遥测消息更新sepcificIndex表消息对象是: {}", JSON.toJSONString(equipmentSpeIndex));
            equipmentSpecificIndexService.updateById(equipmentSpeIndex);

            // 更新设备表指标状态
            iEquipmentSpecificSerivce.updateEquipmentSpecIndexRealtimeData(equipmentSpeIndex);

            equipmentSpecificIndexList.add(equipmentSpeIndex);

            indexStateList.add(createIndexStateVo(equipmentSpeIndex));



            // 火眼数据构造告警指标逻辑
            equipmentSpeIndex = handleTemperatureAlarm(equipmentSpeIndex, iotDatalist);

            boolean alarmFlag = false;
            Map<String, String> messageBodyMap = new HashMap<>();
            //管网压力、泡沫罐信息、水箱液位告警处理
            if (iotDataVO.getKey().equalsIgnoreCase(CAFS_FoamTank_FoamTankLevel) ||
                    FHS_PipePressureDetector_PipePressure.equalsIgnoreCase(iotDataVO.getKey()) ||
                    iotDataVO.getKey().equalsIgnoreCase(CAFS_WaterTank_WaterTankLevel)) {
                alarmFlag = doFoamTankLevel(iotDataVO, equipmentSpeIndex, messageBodyMap);
            }
            //消防水池液位处理
            if (iotDataVO.getKey().equalsIgnoreCase(FHS_FirePoolDevice_WaterLevel) ||
                    iotDataVO.getKey().equalsIgnoreCase(FHS_WirelessliquidDetector_WaterLevel)) {
                alarmFlag = doWaterPoolLevel(iotDataVO, equipmentSpeIndex, messageBodyMap);
            }
            // 遥测数据生成告警事件、日志处理
            if (iotDataVO.getKey().equalsIgnoreCase(CAFS_FoamTank_FoamTankLevel) ||
                    FHS_PipePressureDetector_PipePressure.equalsIgnoreCase(iotDataVO.getKey()) ||
                    iotDataVO.getKey().equalsIgnoreCase(CAFS_WaterTank_WaterTankLevel) ||
                    iotDataVO.getKey().equalsIgnoreCase(FHS_FirePoolDevice_WaterLevel) ||
                    iotDataVO.getKey().equalsIgnoreCase(FHS_WirelessliquidDetector_WaterLevel)) {
                handlingAlarms(equipmentSpeIndex, alarmFlag);
            }

            // 添加指标报告
            saveEquipmentAlarmReportDay(equipmentSpeIndex, alarmFlag);

            // 指标告警处理
            if (equipmentSpeIndex.getIsAlarm() != null && 1 == equipmentSpeIndex.getIsAlarm()) {
                equipmentSpecificAlarms.addAll(createIndexAlarmRecord(equipmentSpeIndex, messageBodyMap));
            }
            // 遥测遥信数据推送云端kafka
            JSONObject jsonObjectXf = new JSONObject();
            jsonObjectXf.put("data_class", "realdata");

            if (equipmentSpeIndex.getIsTrend()) {
                jsonObjectXf.put("data_type", "analog");
            } else {
                jsonObjectXf.put("data_type", "state");
            }

            String date = DateUtils.date2LongStr(new Date());
            jsonObjectXf.put("op_type", "subscribe_emergency");
            JSONObject jsonObjectCondition = new JSONObject();
            jsonObjectCondition.put("station_psr_id", stationCode);
            jsonObjectCondition.put("station_name", stationName);
            jsonObjectCondition.put("data_upload_time", date);
            jsonObjectXf.put("condition", jsonObjectCondition);

            JSONObject jsonObjectData = new JSONObject();
            jsonObjectData.put("psrId", stationCode);
            jsonObjectData.put("astId", equipmentSpeIndex.getSpecificCode());
            jsonObjectData.put("equipType", equipmentSpeIndex.getEquipmentCode());
            jsonObjectData.put("name", equipmentSpeIndex.getEquipmentSpecificName() + "-" + equipmentSpeIndex.getEquipmentSpecificIndexName());
            if (value.equals("true")) {
                jsonObjectData.put("value", "1");
            } else if (value.equals("false")) {
                jsonObjectData.put("value", "0");
            } else {
                jsonObjectData.put("value", value);
            }
            jsonObjectData.put("measurementType", null == equipmentSpeIndex.getEquipmentIndexKey() ? "" : equipmentSpeIndex.getEquipmentIndexKey());
            jsonObjectData.put("dateTime", date);
            jsonObjectData.put("quality", "0"); // 量测质量码：0 有效，1 无效
            List<JSONObject> jsonObjects = Collections.singletonList(jsonObjectData);

            jsonObjectXf.put("data", jsonObjects);

            // 遥测
            if (!isOpenTelemetering && equipmentSpeIndex.getIsTrend()) {

            } else {
                try {
                    emqKeeper.getMqttClient().publish("emq.xf.created", jsonObjectXf.toString().getBytes(), 1, false);
                    log.info("遥测遥信数据推送云端kafka成功");
                } catch (MqttException e) {
                    log.error("遥测遥信数据推送云端kafka失败=====>" + e.getMessage());
                }
            }

            // 报警数据保存
            List<EquipmentSpecificAlarmLog> alarmLogs = new ArrayList<>();
            if (!ObjectUtils.isEmpty(equipmentSpecificAlarms)) {
                equipmentSpecificAlarmService.saveOrUpdateBatch(equipmentSpecificAlarms);
            }

            // 需要在事务提交之后，否则事务隔离查询不出数据
            TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
                @Override
                public void afterCommit() {
                    equipmentSpecificAlarms.forEach(action -> {
                        if (AlarmStatusEnum.BJ.getCode() == action.getStatus()) {
                            alarmLogs.add(addEquipAlarmLogRecord(action));
                            if (ValidationUtil.isEmpty(action.getAlamContent())) {
                                action.setAlamContent(action.getEquipmentSpecificName() + action.getEquipmentSpecificIndexName());
                            }
                            mqttSendGateway.sendToMqtt(TopicEnum.EQDQR.getTopic(), JSONArray.toJSON(action).toString());
                            bool = Boolean.FALSE;
                        } else {
                            alarmLogs.addAll(upAlarmLogStatus(action.getIotCode(), action.getEquipmentSpecificIndexKey(), action.getTraceId(),
                                    equipmentSpecificAlarmLogService, false, false, ""));
                            mqttSendGateway.sendToMqtt(TopicEnum.EQYQR.getTopic(), JSONArray.toJSON(action).toString());
                            bool = Boolean.TRUE;
                        }
                    });

                    // // 告警消息推送
                    if (!ObjectUtils.isEmpty(equipmentSpecificAlarms)) {
                        publishDataToDCCenterPage(equipmentSpecificIndexList);
                    }

					//推送数据到组态大屏(消防系统)
					pushDataToIntegrationPage(equipmentSpecificIndexList);
                    // 四横八纵遥测信号信息列表刷新
                    publishNormalIndexValueToPage(equipmentSpecificIndexList);
                    if ("zd".equals(system)) {
                        System.out.println("站端系统----------------");

                        // 向预控系统发送消息
                        sendEquipSpecIndexToAutosysTopic(equipmentSpecificIndexList);

                        // 首页性能指标数据订阅
                        mqttSendGateway.sendToMqtt(indexTopic, JSON.toJSONString(indexStateList));

                        // 组态大屏消息推送，设备表实时指标修改
                        intePageSysDataRefresh(equipmentSpecificIndexList, topicEntity);

                        // 数字换流站同步指标修改
                        syncSpecificIndexsToGS(equipmentSpecificIndexList);

                        // 则更新拓扑节点数据及告警状态
                        updateNodeDateByEquipId(equipmentSpecificIndexList);

                        // 向画布推送
                        publishDataToCanvas(equipmentSpecificIndexList);

                        // 向其他系统推送报警
                        try {
                            equipmentAlarmLogsToOtherSystems(alarmLogs);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }

                        if (equipmentSpecificVo.getEcode() != null) {
                            String ecode = equipmentSpecificVo.getEcode();
                            boolean flag = false;

                            //消防泵
                            String[] strings = pumpCodes.split(",");
                            for (String string : strings) {
                                if (ecode.startsWith(string)) {
                                    //通知>消防应急预案
                                    topicEntity.setType("xfb");
                                    mqttSendGateway.sendToMqtt(emergencyDisposalIndicators, JSONObject.toJSONString(topicEntity));
                                    flag = true;
                                    break;
                                }
                            }

                            // 消防炮
                            String[] stringxfp = monitorCodes.split(",");
                            if (!flag) {
                                for (String string1 : stringxfp) {
                                    if (ecode.startsWith(string1)) {
                                        //通知>消防应急预案
                                        topicEntity.setType("xfp");
                                        mqttSendGateway.sendToMqtt(emergencyDisposalIndicators, JSONObject.toJSONString(topicEntity));
                                        flag = true;
                                        break;
                                    }
                                }
                            }
                            //消防水源
                            if (!flag) {
                                List<Map> lit = iEquipmentSpecificSerivce.getWater(equipmentSpecificVo.getId());
                                if (lit != null && !lit.isEmpty()) {
                                    topicEntity.setType("xfsy");
                                    mqttSendGateway.sendToMqtt(emergencyDisposalIndicators, JSONObject.toJSONString(topicEntity));
                                }
                            }
                        }
                    }
                }
            });
        }
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void handlerMqttStationMessage(String topic, String message) {
        log.info("接收到韶山Kafka消息: {}", message);
        TopicEntityVo topicEntity = new TopicEntityVo();
        topicEntity.setTopic(topic);
        topicEntity.setMessage(message);

        List<IotDataVO> iotDatalist = new ArrayList<>();
        List<EquipmentSpecificIndex> equipmentSpecificIndexList = new ArrayList<>();
        List<EquipmentSpecificAlarm> equipmentSpecificAlarms = new ArrayList<>();
        List<IndexStateVo> indexStateList = new ArrayList<>();
        JSONObject messageObj = JSONObject.parseObject(message);
        JSONObject data = messageObj.getJSONObject("data");
        if (!ObjectUtils.isEmpty(data)) {
            message = JSON.toJSONString(data);
//            log.info("Kafka消息取data数据: {}", data);
        }

//        message = "{\"timeStamp\":\"2024-05-15 15:29:01.883\",\"warns\":[{\"deviceId\":\"sxxfb30000001112967078\",\"eventstatus\":\"出现\",\"pointId\":\"LS.S2.RCS9794-ZKL.-.绍兴消防泵.2号稳压泵运行\",\"time\":\"2024-05-15 15:29:01.883\"}]}";
        SShanStationMessage sShanStationMessage = JSON.parseObject(message, SShanStationMessage.class);
        log.info("转换message数据: {}", sShanStationMessage);
        Map<Object, Object> equipmentIndexKeyMap = redisUtils.hmget(RedisKey.EQUIP_INDEX_ADDRESS_KEY_STATION);
//        log.info("Redis取数据key==equip_index_address_key_station: {}", equipmentIndexKeyMap);
        for (SShanMessage m : sShanStationMessage.getWarns()) {
            String value;
            if (message.contains("eventtextL1")) {
                // 此通道目前专供韶山告警消息value赋值，没必要用配置文件多字段进行逐个判断，所以暂时不修改
                value = m.getEventTextL1().contains("出现") ? TRUE : FALSE;
            } else {
                value = String.valueOf(BeanUtil.judgePropertyContainsChar(m.getEventstatus(), specialChars));
            }

            if (equipmentIndexKeyMap.get(m.getPointId()) != null) {
                EquipmentSpecificIndex equipmentSpeIndex = equipmentSpecificIndexService.getEquipmentSpeIndexByAddress(null, m.getPointId(), null);
//                log.info("查询equip_specific_index数据=======：", JSON.toJSONString(equipmentSpeIndex.toString()));
                if (equipmentSpeIndex == null) {
                    return;
                }
                equipmentSpeIndex.setValue(value);
                equipmentSpeIndex.setValueLabel(valueTranslate(value, equipmentSpeIndex.getValueEnum()));
                equipmentSpeIndex.setEquipmentType(topicEntity.getType());
                equipmentSpeIndex.setUpdateDate(new Date());
                equipmentSpeIndex.setTimeStamp(sShanStationMessage.getTimestamp());
                equipmentSpeIndex.setUUID(UUIDUtils.getUUID());

                IotDataVO iotDataVO = new IotDataVO();
                iotDataVO.setKey(equipmentSpeIndex.getNameKey());
                iotDataVO.setValue(value);
                iotDatalist.add(iotDataVO);

                QueryWrapper<EquipmentSpecific> queryWrapper = new QueryWrapper<>();
                queryWrapper.eq("id", equipmentSpeIndex.getEquipmentSpecificId());
                EquipmentSpecific equipmentSpecific = iEquipmentSpecificSerivce.getOne(queryWrapper);
                if (equipmentSpecific == null) {
                    return;
                }
                String iotCode = equipmentSpecific.getIotCode();
                StringBuilder endIndex = new StringBuilder(iotCode).insert(8, '/');
                String iotTopic = "influxdb/" + endIndex;
                JSONObject msg = new JSONObject();
                msg.put(equipmentSpeIndex.getEquipmentIndexKey(), value);
                mqttSendGateway.sendToMqtt(iotTopic, JSON.toJSONString(msg));

                List<EquipmentSpecificVo> eqIotCodeList = iEquipmentSpecificSerivce.getEquipAndCarIotcodeByIotcode(iotCode);

                if (eqIotCodeList.isEmpty()) {
                    log.info("该数据{}不存在！", iotCode);
                    return;
                }
                if (eqIotCodeList.size() > 1) {
                    log.info("有重复的{}数据！", iotCode);
                }
                EquipmentSpecificVo equipmentSpecificVo = eqIotCodeList.get(0);
                topicEntity.setType(equipmentSpecificVo.getType());
                topicEntity.setCode(equipmentSpecificVo.getCode());

                //es存储数据
                eSeqService.saveESEquiplistSpecificBySystemESVO(equipmentSpeIndex, String.valueOf(equipmentSpecificVo.getSystemId()), equipmentSpecificVo.getSystemName());
                //更新装备性能指标
                equipmentSpecificIndexService.updateById(equipmentSpeIndex);

                // 更新设备表指标状态
                iEquipmentSpecificSerivce.updateEquipmentSpecIndexRealtimeData(equipmentSpeIndex);

                equipmentSpecificIndexList.add(equipmentSpeIndex);

                indexStateList.add(createIndexStateVo(equipmentSpeIndex));



                // 火眼数据构造告警指标逻辑
                equipmentSpeIndex = handleTemperatureAlarm(equipmentSpeIndex, iotDatalist);

                boolean alarmFlag = false;
                Map<String, String> messageBodyMap = new HashMap<>();
                //管网压力、泡沫罐信息、水箱液位告警处理
                if (iotDataVO.getKey().equalsIgnoreCase(CAFS_FoamTank_FoamTankLevel) ||
                        FHS_PipePressureDetector_PipePressure.equalsIgnoreCase(iotDataVO.getKey()) ||
                        iotDataVO.getKey().equalsIgnoreCase(CAFS_WaterTank_WaterTankLevel)) {
                    alarmFlag = doFoamTankLevel(iotDataVO, equipmentSpeIndex, messageBodyMap);
                }
                //消防水池液位处理
                if (iotDataVO.getKey().equalsIgnoreCase(FHS_FirePoolDevice_WaterLevel) ||
                        iotDataVO.getKey().equalsIgnoreCase(FHS_WirelessliquidDetector_WaterLevel)) {
                    alarmFlag = doWaterPoolLevel(iotDataVO, equipmentSpeIndex, messageBodyMap);
                }
                // 遥测数据生成告警事件、日志处理
                if (iotDataVO.getKey().equalsIgnoreCase(CAFS_FoamTank_FoamTankLevel) ||
                        FHS_PipePressureDetector_PipePressure.equalsIgnoreCase(iotDataVO.getKey()) ||
                        iotDataVO.getKey().equalsIgnoreCase(CAFS_WaterTank_WaterTankLevel) ||
                        iotDataVO.getKey().equalsIgnoreCase(FHS_FirePoolDevice_WaterLevel) ||
                        iotDataVO.getKey().equalsIgnoreCase(FHS_WirelessliquidDetector_WaterLevel)) {
                    handlingAlarms(equipmentSpeIndex, alarmFlag);
                }

                // 添加指标报告
                saveEquipmentAlarmReportDay(equipmentSpeIndex, alarmFlag);

                // 指标告警处理
                if (equipmentSpeIndex.getIsAlarm() != null && 1 == equipmentSpeIndex.getIsAlarm()) {
                    equipmentSpecificAlarms.addAll(createIndexAlarmRecord(equipmentSpeIndex, messageBodyMap));
                }
                // 遥测遥信数据推送云端kafka
                JSONObject jsonObjectXf = new JSONObject();
                jsonObjectXf.put("data_class", "realdata");

                if (equipmentSpeIndex.getIsTrend()) {
                    jsonObjectXf.put("data_type", "analog");
                } else {
                    jsonObjectXf.put("data_type", "state");
                }

                String date = DateUtils.date2LongStr(new Date());
                jsonObjectXf.put("op_type", "subscribe_emergency");
                JSONObject jsonObjectCondition = new JSONObject();
                jsonObjectCondition.put("station_psr_id", stationCode);
                jsonObjectCondition.put("station_name", stationName);
                jsonObjectCondition.put("data_upload_time", date);
                jsonObjectXf.put("condition", jsonObjectCondition);

                JSONObject jsonObjectData = new JSONObject();
                jsonObjectData.put("psrId", stationCode);
                jsonObjectData.put("astId", equipmentSpeIndex.getSpecificCode());
                jsonObjectData.put("equipType", equipmentSpeIndex.getEquipmentCode());
                jsonObjectData.put("name", equipmentSpeIndex.getEquipmentSpecificName() + "-" + equipmentSpeIndex.getEquipmentSpecificIndexName());
                if (value.equals("true")) {
                    jsonObjectData.put("value", "1");
                } else {
                    jsonObjectData.put("value", "0");
                }
                jsonObjectData.put("measurementType", null == equipmentSpeIndex.getEquipmentIndexKey() ? "" : equipmentSpeIndex.getEquipmentIndexKey());
                jsonObjectData.put("dateTime", date);
                jsonObjectData.put("quality", "0"); // 量测质量码：0 有效，1 无效
                List<JSONObject> jsonObjects = Collections.singletonList(jsonObjectData);

                jsonObjectXf.put("data", jsonObjects);

                // 遥测
                if (!isOpenTelemetering && equipmentSpeIndex.getIsTrend()) {

                } else {
                    try {
                        emqKeeper.getMqttClient().publish("emq.xf.created", jsonObjectXf.toString().getBytes(), 1, false);
                        log.info("告警事件数据推送云端kafka成功");
                    } catch (MqttException e) {
                        log.error("告警事件数据推送云端kafka失败=====>" + e.getMessage());
                        e.printStackTrace();
                    }
                }

                // 报警数据保存
                List<EquipmentSpecificAlarmLog> alarmLogs = new ArrayList<>();
                if (!ObjectUtils.isEmpty(equipmentSpecificAlarms)) {
                    equipmentSpecificAlarmService.saveOrUpdateBatch(equipmentSpecificAlarms);
                }

                // 需要在事务提交之后，否则事务隔离查询不出数据
                TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
                    @Override
                    public void afterCommit() {
                        equipmentSpecificAlarms.forEach(action -> {
                            if (AlarmStatusEnum.BJ.getCode() == action.getStatus()) {
                                alarmLogs.add(addEquipAlarmLogRecord(action));
                                if (ValidationUtil.isEmpty(action.getAlamContent())) {
                                    action.setAlamContent(action.getEquipmentSpecificName() + action.getEquipmentSpecificIndexName());
                                }
                                mqttSendGateway.sendToMqtt(TopicEnum.EQDQR.getTopic(), JSONArray.toJSON(action).toString());
                                bool = Boolean.FALSE;
                            } else {
                                alarmLogs.addAll(upAlarmLogStatus(action.getIotCode(), action.getEquipmentSpecificIndexKey(), action.getTraceId(),
                                        equipmentSpecificAlarmLogService, false, false, ""));
                                mqttSendGateway.sendToMqtt(TopicEnum.EQYQR.getTopic(), JSONArray.toJSON(action).toString());
                                bool = Boolean.TRUE;
                            }
                        });

                        // // 告警消息推送
                        if (!ObjectUtils.isEmpty(equipmentSpecificAlarms)) {
                            publishDataToDCCenterPage(equipmentSpecificIndexList);
                        }

                        //推送数据到组态大屏(消防系统)
    					pushDataToIntegrationPage(equipmentSpecificIndexList);
                        // 四横八纵遥测信号信息列表刷新
                        publishNormalIndexValueToPage(equipmentSpecificIndexList);
                        if ("zd".equals(system)) {
                            System.out.println("站端系统----------------");

                            // 向预控系统发送消息
                            sendEquipSpecIndexToAutosysTopic(equipmentSpecificIndexList);

                            // 首页性能指标数据订阅
                            mqttSendGateway.sendToMqtt(indexTopic, JSON.toJSONString(indexStateList));

                            // 组态大屏消息推送，设备表实时指标修改
                            intePageSysDataRefresh(equipmentSpecificIndexList, topicEntity);

                            // 数字换流站同步指标修改
                            syncSpecificIndexsToGS(equipmentSpecificIndexList);

                            // 则更新拓扑节点数据及告警状态
                            updateNodeDateByEquipId(equipmentSpecificIndexList);

                            // 向画布推送
                            publishDataToCanvas(equipmentSpecificIndexList);

                            // 向其他系统推送报警
                            try {
                                equipmentAlarmLogsToOtherSystems(alarmLogs);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }

                            if (equipmentSpecificVo.getEcode() != null) {
                                String ecode = equipmentSpecificVo.getEcode();
                                boolean flag = false;

                                //消防泵
                                String[] strings = pumpCodes.split(",");
                                for (String string : strings) {
                                    if (ecode.startsWith(string)) {
                                        //通知>消防应急预案
                                        topicEntity.setType("xfb");
                                        mqttSendGateway.sendToMqtt(emergencyDisposalIndicators, JSONObject.toJSONString(topicEntity));
                                        flag = true;
                                        break;
                                    }
                                }

                                // 消防炮
                                String[] stringxfp = monitorCodes.split(",");
                                if (!flag) {
                                    for (String string1 : stringxfp) {
                                        if (ecode.startsWith(string1)) {
                                            //通知>消防应急预案
                                            topicEntity.setType("xfp");
                                            mqttSendGateway.sendToMqtt(emergencyDisposalIndicators, JSONObject.toJSONString(topicEntity));
                                            flag = true;
                                            break;
                                        }
                                    }
                                }
                                //消防水源
                                if (!flag) {
                                    List<Map> lit = iEquipmentSpecificSerivce.getWater(equipmentSpecificVo.getId());
                                    if (lit != null && !lit.isEmpty()) {
                                        topicEntity.setType("xfsy");
                                        mqttSendGateway.sendToMqtt(emergencyDisposalIndicators, JSONObject.toJSONString(topicEntity));
                                    }
                                }
                            }
                        }
                    }
                });
            }
        }
    }


    /**
     * 物联数据处理
     *
     * @param topicEntity
     * @param iotDatalist
     */
    public void realTimeDateProcessing(TopicEntityVo topicEntity, List<IotDataVO> iotDatalist, EquipmentSpecificVo vo) {
        String iotCode = topicEntity.getIotCode();
        if (EquipAndCarEnum.equip.type.equals(topicEntity.getType())) {
            List<EquipmentSpecificIndex> indexList = equipmentSpecificIndexService
                    .getEquipmentSpeIndexBySpeIotCode(iotCode);
            if (ObjectUtils.isEmpty(indexList)) {
                return;
            }
            equipRealTimeDate(iotDatalist, indexList, topicEntity, vo);
            String bizOrgCode = indexList.get(0).getBizOrgCode();
            //  redis缓存指定指标、指定时长物联数据
            pressurePumpService.saveDataToRedis(iotDatalist, iotCode, bizOrgCode);
        } else {
            List<CarProperty> carProperties = carPropertyService.getCarPropListByIotCode(iotCode);
            if (ObjectUtils.isEmpty(carProperties)) {
                return;
            }
            carRealTimeDate(iotDatalist, carProperties, topicEntity);
        }
    }

    /**
     * 装备实时数据处理
     *
     * @param iotDatalist
     * @param indexList
     * @param topicEntity
     */
    public void equipRealTimeDate(List<IotDataVO> iotDatalist, List<EquipmentSpecificIndex> indexList,
                                  TopicEntityVo topicEntity, EquipmentSpecificVo vo) {
        List<EquipmentSpecificIndex> equipmentSpecificIndexList = new ArrayList<>();
        List<EquipmentSpecificAlarm> equipmentSpecificAlarms = new ArrayList<>();
        List<IndexStateVo> indexStateList = new ArrayList<>();
        // 存储温度数据至内存中（火眼）
        iotDataListToCacheMap(iotDatalist);
        iotDatalist.forEach(iotDataVO -> {
            boolean alarmFlag = false;

            for (EquipmentSpecificIndex equipmentSpecificIndex : indexList) {
                if (!ObjectUtils.isEmpty(equipmentSpecificIndex.getNameKey())
                        && equipmentSpecificIndex.getNameKey().equalsIgnoreCase(iotDataVO.getKey())) {
                    EquipmentSpecificIndex equipmentSpeIndex = new EquipmentSpecificIndex();
                    BeanUtils.copyProperties(equipmentSpecificIndex, equipmentSpeIndex);
                    String value = iotDataVO.getValue().toString();
                    String traceId = iotDataVO.getTraceId();
                    equipmentSpeIndex.setValue(value);
                    equipmentSpeIndex.setUnit(equipmentSpecificIndex.getIndexUnitName());
                    equipmentSpeIndex.setTraceId(traceId);
                    equipmentSpeIndex.setValueLabel(valueTranslate(value, equipmentSpecificIndex.getValueEnum()));
                    equipmentSpeIndex.setUpdateDate(new Date());
                    equipmentSpeIndex.setEquipmentType(topicEntity.getType());
                    equipmentSpeIndex.setUUID(UUIDUtils.getUUID());

//                    if ("zd".equals(system)) {
//                        // 收到信号触发规则恢复码判断
//                        List<Map<String, String>> list = new ArrayList<>();
//                        Map<String, String> map = new HashMap<>(2);
//                        map.put("sourceId", String.valueOf(equipmentSpecificIndex.getEquipmentSpecificId()));
//                        map.put("indexValue", value);
//                        map.put("specificIndexId", String.valueOf(equipmentSpecificIndex.getId()));
//                        list.add(map);
//                        EquipQrDateDto equipQrDateDto = new EquipQrDateDto();
//                        equipQrDateDto.setContrast("indexValue");
//                        equipQrDateDto.setSource("equip");
//                        equipQrDateDto.setType(equipmentSpecificIndex.getTypeCode());
//                        equipQrDateDto.setData(list);
//                        log.error("调用规则JSON对象：{}", JSONObject.toJSONString(equipQrDateDto));
//                        try {
//                            ruleTrigger.publish(equipQrDateDto, "中心配置赋码规则/update-qr-code", null);
//                        } catch (Exception e) {
//                            log.error("调用规则失败: {}", e.getMessage());
//                        }
//                    }

                    //es存储数据
                    eSeqService.saveESEquiplistSpecificBySystemESVO(equipmentSpeIndex, String.valueOf(vo.getSystemId()), vo.getSystemName());

                    equipmentSpecificIndexService.updateById(equipmentSpeIndex);

                    // 更新设备表指标状态
                    iEquipmentSpecificSerivce.updateEquipmentSpecIndexRealtimeData(equipmentSpeIndex);

                    equipmentSpecificIndexList.add(equipmentSpeIndex);

                    indexStateList.add(createIndexStateVo(equipmentSpeIndex));

                    // 推送数据到组态大屏
                    pushDataToIntegrationPage(equipmentSpecificIndexList);



                    // 火眼数据构造告警指标逻辑
                    equipmentSpecificIndex = handleTemperatureAlarm(equipmentSpeIndex, iotDatalist);
                    // 告警日志表消息内容
                    Map<String, String> messageBodyMap = new HashMap<>();
                    //管网压力、泡沫罐信息、水箱液位告警处理
                    if (iotDataVO.getKey().equalsIgnoreCase(CAFS_FoamTank_FoamTankLevel) ||
                            FHS_PipePressureDetector_PipePressure.equalsIgnoreCase(iotDataVO.getKey()) ||
                            iotDataVO.getKey().equalsIgnoreCase(CAFS_WaterTank_WaterTankLevel)) {
                        alarmFlag = doFoamTankLevel(iotDataVO, equipmentSpecificIndex, messageBodyMap);
                    }
                    //消防水池液位处理
                    if (iotDataVO.getKey().equalsIgnoreCase(FHS_FirePoolDevice_WaterLevel) ||
                            iotDataVO.getKey().equalsIgnoreCase(FHS_WirelessliquidDetector_WaterLevel)) {
                        alarmFlag = doWaterPoolLevel(iotDataVO, equipmentSpecificIndex, messageBodyMap);
                    }
                    // 遥测数据生成告警事件、日志处理
                    if (iotDataVO.getKey().equalsIgnoreCase(CAFS_FoamTank_FoamTankLevel) ||
                            FHS_PipePressureDetector_PipePressure.equalsIgnoreCase(iotDataVO.getKey()) ||
                            iotDataVO.getKey().equalsIgnoreCase(CAFS_WaterTank_WaterTankLevel) ||
                            iotDataVO.getKey().equalsIgnoreCase(FHS_FirePoolDevice_WaterLevel) ||
                            iotDataVO.getKey().equalsIgnoreCase(FHS_WirelessliquidDetector_WaterLevel)) {
                        handlingAlarms(equipmentSpecificIndex, alarmFlag);
                    }

                    // 添加指标报告
                    saveEquipmentAlarmReportDay(equipmentSpeIndex, alarmFlag);

                    // 指标告警处理
                    if (equipmentSpecificIndex.getIsAlarm() != null && 1 == equipmentSpecificIndex.getIsAlarm() && !equipmentSpecificIndex.getEquipmentIndexKey().equals(pressurePumpStart)) {
                        equipmentSpecificAlarms.addAll(createIndexAlarmRecord(equipmentSpecificIndex, messageBodyMap));
                    }
                    // 遥测遥信数据推送云端kafka
                    JSONObject jsonObjectXf = new JSONObject();
                    jsonObjectXf.put("data_class", "realdata");

                    if (equipmentSpeIndex.getIsTrend()) {
                        jsonObjectXf.put("data_type", "analog");
                    } else {
                        jsonObjectXf.put("data_type", "state");
                    }

                    String date = DateUtils.date2LongStr(new Date());
                    jsonObjectXf.put("op_type", "subscribe_emergency");
                    JSONObject jsonObjectCondition = new JSONObject();
                    jsonObjectCondition.put("station_psr_id", stationCode);
                    jsonObjectCondition.put("station_name", stationName);
                    jsonObjectCondition.put("data_upload_time", date);
                    jsonObjectXf.put("condition", jsonObjectCondition);

                    JSONObject jsonObjectData = new JSONObject();
                    jsonObjectData.put("psrId", stationCode);
                    jsonObjectData.put("astId", equipmentSpecificIndex.getSpecificCode());
                    jsonObjectData.put("equipType", equipmentSpecificIndex.getEquipmentCode());
                    jsonObjectData.put("name", equipmentSpecificIndex.getEquipmentSpecificName() + "-" + equipmentSpecificIndex.getEquipmentSpecificIndexName());
                    if (value.equals("true")) {
                        jsonObjectData.put("value", "1");
                    } else if (value.equals("false")) {
                        jsonObjectData.put("value", "0");
                    } else {
                        jsonObjectData.put("value", value);
                    }
                    jsonObjectData.put("measurementType", null == equipmentSpecificIndex.getEquipmentIndexKey() ? "" : equipmentSpecificIndex.getEquipmentIndexKey());
                    jsonObjectData.put("dateTime", date);
                    jsonObjectData.put("quality", "0"); // 量测质量码：0 有效，1 无效
                    List<JSONObject> jsonObjects = Collections.singletonList(jsonObjectData);

                    jsonObjectXf.put("data", jsonObjects);

                    // 遥测
                    if (!isOpenTelemetering && equipmentSpeIndex.getIsTrend()) {

                    } else {
                        try {
                            emqKeeper.getMqttClient().publish("emq.xf.created", jsonObjectXf.toString().getBytes(), 1, false);
                            log.info("遥测遥信数据推送云端kafka成功");
                        } catch (MqttException e) {
                            log.error("遥测遥信数据推送云端kafka失败=====>" + e.getMessage());
                            e.printStackTrace();
                        }
                    }

                    //根据指标数据处理预案按钮
                    String json = null;
                    try {
                        json = IOUtils.toString(systemStateIndex.getInputStream(), java.lang.String.valueOf(StandardCharsets.UTF_8));
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    List<Map> mapList = JSONObject.parseArray(json, Map.class);
                    List<EquipIndexDto> indexDtos = new ArrayList<>();
                    mapList.forEach(x -> {
                        List<EquipIndexDto> list = JSONArray.parseArray(String.valueOf(x.get("indexVo")), EquipIndexDto.class);
                        indexDtos.addAll(list);
                    });
                    List<EquipIndexDto> list = indexDtos.stream().filter(x -> iotDataVO.getKey().equalsIgnoreCase(String.valueOf(x.getIndexKey()))).collect(Collectors.toList());
                    if(0 < list.size()){
                        EquipIndexDto indexDto = new EquipIndexDto();
                        indexDto.setIndexKey(iotDataVO.getKey());
                        indexDto.setIndexValue(value);
                        systemStateExec(indexDto);
                    }
                }
            }
        });

        // 报警数据保存
        List<EquipmentSpecificAlarmLog> alarmLogs = new ArrayList<>();
        if (!ObjectUtils.isEmpty(equipmentSpecificAlarms)) {
            equipmentSpecificAlarmService.saveOrUpdateBatch(equipmentSpecificAlarms);
        }
        // 需要在事务提交之后，否则事务隔离查询不出数据
        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
            @Override
            public void afterCommit() {
                iotDatalist.forEach(iotDataVO -> {
                    String indexKey = iotDataVO.getKey();
                    String indexValue = iotDataVO.getValue().toString();
                    // 稳压泵启停信号处理
                    if (indexKey.equals(pressurePumpStart)) {
                        pressurePump(indexKey, indexValue, iotDatalist, topicEntity);
                    }
                });

                equipmentSpecificAlarms.forEach(action -> {
                    if (AlarmStatusEnum.BJ.getCode() == action.getStatus()) {
                        alarmLogs.add(addEquipAlarmLogRecord(action));
                        if (ValidationUtil.isEmpty(action.getAlamContent())) {
                            action.setAlamContent(action.getEquipmentSpecificName() + action.getEquipmentSpecificIndexName());
                        }
                        mqttSendGateway.sendToMqtt(TopicEnum.EQDQR.getTopic(), JSONArray.toJSON(action).toString());
                        bool = Boolean.FALSE;
                    } else {
                        alarmLogs.addAll(upAlarmLogStatus(action.getIotCode(), action.getEquipmentSpecificIndexKey(), action.getTraceId(),
                                equipmentSpecificAlarmLogService, false, false, ""));
                        mqttSendGateway.sendToMqtt(TopicEnum.EQYQR.getTopic(), JSONArray.toJSON(action).toString());
                        bool = Boolean.TRUE;
                    }
                });

                // 告警消息推送
                 if (!ObjectUtils.isEmpty(equipmentSpecificAlarms)) {
                    publishDataToDCCenterPage(equipmentSpecificIndexList);
                }

                // 四横八纵遥测信号信息列表刷新
                publishNormalIndexValueToPage(equipmentSpecificIndexList);

                // 触发风险---> 站端发送消息到Message服务
                publishDataToMessage(equipmentSpecificIndexList, isAlarm);

                if ("zd".equals(system)) {
                    System.out.println("站端系统----------------");

                    // 向预控系统发送消息
                    sendEquipSpecIndexToAutosysTopic(equipmentSpecificIndexList);

                    // 首页性能指标数据订阅
                    mqttSendGateway.sendToMqtt(indexTopic, JSON.toJSONString(indexStateList));

                    // 组态大屏消息推送，设备表实时指标修改
                    intePageSysDataRefresh(equipmentSpecificIndexList, topicEntity);

                    // 数字换流站同步指标修改
                    syncSpecificIndexsToGS(equipmentSpecificIndexList);

                    // 则更新拓扑节点数据及告警状态
                    updateNodeDateByEquipId(equipmentSpecificIndexList);

                    // 向画布推送
                    publishDataToCanvas(equipmentSpecificIndexList);

                    // 向其他系统推送报警
                    try {
                        equipmentAlarmLogsToOtherSystems(alarmLogs);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }

                    if (vo.getEcode() != null) {
                        String ecode = vo.getEcode();
                        boolean flag = false;

                        //消防泵
                        String[] strings = pumpCodes.split(",");
                        for (String string : strings) {
                            if (ecode.startsWith(string)) {
                                //通知>消防应急预案
                                topicEntity.setType("xfb");
                                mqttSendGateway.sendToMqtt(emergencyDisposalIndicators, JSONObject.toJSONString(topicEntity));
                                flag = true;
                                break;
                            }
                        }

                        // 消防炮
                        String[] stringxfp = monitorCodes.split(",");
                        if (!flag) {
                            for (String string1 : stringxfp) {
                                if (ecode.startsWith(string1)) {
                                    //通知>消防应急预案
                                    topicEntity.setType("xfp");
                                    mqttSendGateway.sendToMqtt(emergencyDisposalIndicators, JSONObject.toJSONString(topicEntity));
                                    flag = true;
                                    break;
                                }
                            }
                        }
                        //消防水源
                        if (!flag) {
                            List<Map> lit = iEquipmentSpecificSerivce.getWater(vo.getId());
                            if (lit != null && lit.size() > 0) {
                                topicEntity.setType("xfsy");
                                mqttSendGateway.sendToMqtt(emergencyDisposalIndicators, JSONObject.toJSONString(topicEntity));
                            }
                        }
                    }
                }
            }
        });

    }

    // 触发规则调用，修改消防系统运行状态
    private void systemStateExec(EquipIndexDto indexDto) {
        String autoExecPath = "换流站消防专项预案/autoExecuteSystemState";
        try {
//            RequestContext.setAppKey(amosAuth.getAppKey());
//            RequestContext.setProduct(amosAuth.getProduct());
//            RequestContext.setToken(amosAuth.getToken());
            log.info("装备指标触发规则>>>>>>>>>>>>>>>>>>>>>" + JSON.toJSONString(indexDto) + " >>>>>>>>>>>>>>" + JSON.toJSONString(autoExecPath));
            ruleTrigger.publish(indexDto, autoExecPath, new String[] {});
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void handlingAlarms(EquipmentSpecificIndex equipmentSpecificIndex, boolean alarmFlag) {
        equipmentSpecificIndex.setIsAlarm(1);
        equipmentSpecificIndex.setValue(String.valueOf(alarmFlag));
    }

    /**
     * 泡沫罐  或 者管网压力  消息发送
     *
     * @param iotDataVO              iotDataVO
     * @param equipmentSpecificIndex equipmentSpecificIndex
     */
    private boolean doFoamTankLevel(IotDataVO iotDataVO, EquipmentSpecificIndex equipmentSpecificIndex, Map<String, String> messageBody) {
        boolean alarmFlag = false;
        MessageModel model = new MessageModel();
        Map<String, Object> map = new HashMap<>();
        if (iotDataVO.getKey().equalsIgnoreCase(CAFS_FoamTank_FoamTankLevel) || iotDataVO.getKey().equalsIgnoreCase(CAFS_WaterTank_WaterTankLevel)) {
            map = fireFightingSystemMapper.getFoamTankLevel(equipmentSpecificIndex.getEquipmentSpecificId());
        } else {
            map = fireFightingSystemMapper.getPipeNetwork(equipmentSpecificIndex.getEquipmentSpecificId());
        }
        BigDecimal minValue = getBigDecimal(map.get("minValues"));
        BigDecimal maxValue = getBigDecimal(map.get("maxValues"));
        BigDecimal nowValue = getBigDecimal(iotDataVO.getValue());
        BigDecimal checkValue = checkUnit(equipmentSpecificIndex.getEquipmentSpecificId(), iotDataVO.getKey().toLowerCase(), nowValue);
        if (!ObjectUtils.isEmpty(checkValue)) {
            nowValue = checkValue;
        }
        if (nowValue.compareTo(minValue) < 0 || nowValue.compareTo(maxValue) > 0) {
            String body = "";
            if (nowValue.compareTo(minValue) < 0) {
                body = "当前数值 " + nowValue + " 低于最低报警阈值 " + minValue;
            } else if (nowValue.compareTo(maxValue) > 0) {
                body = "当前数值 " + nowValue + " 超过最高报警阈值 " + maxValue;
            }
            messageBody.put("messageBody", equipmentSpecificIndex.getEquipmentSpecificName() + "-" + body);
            String bodyMain = String.format("%s，- 当前数值%s，%s ，请及时查看处理。",
                    equipmentSpecificIndex.getEquipmentSpecificName() + "-" + equipmentSpecificIndex.getLocation(),
                    nowValue,
                    nowValue.compareTo(minValue) < 0 ? "低于最低报警阈值" + minValue : "超过最高报警阈值" + maxValue
            );
            model.setTitle("模拟量提醒");
            model.setBody(bodyMain);
            model.setMsgType("FoamTankOrPipeNetwork");
            model.setSendTime(new Date());
            model.setIsSendWeb(true);
            model.setCategory(1);
            model.setRelationId(equipmentSpecificIndex.getEquipmentSpecificId().toString());
            model.setIsSendApp(false);
            model.setTerminal("WEB");
            model.setRecivers(Collections.singletonList("system"));
            Map<String, String> ext = new HashMap<>();
            ext.put("content", body);
            ext.put("type", "模拟量超阈值提醒");
            ext.put("name", equipmentSpecificIndex.getEquipmentSpecificName());
            ext.put("time", new SimpleDateFormat(DateUtils.DATE_TIME_PATTERN).format(new Date()));
            model.setExtras(ext);
//            try {
//                Token token = remoteSecurityService.getServerToken();
//                systemctlFeign.create(token.getAppKey(), token.getProduct(), token.getToke(), model);
//            } catch (Exception e) {
//                log.error("调用平台出错！");
//            }
//            log.info(String.format("调用平台消息服务成功：%s", JSON.toJSONString(model)));
            alarmFlag = true;
        }
        return alarmFlag;
    }

    /**
     * 消防水池、工业水池和消防水箱  消息发送
     *
     * @param iotDataVO              iotDataVO
     * @param equipmentSpecificIndex equipmentSpecificIndex
     */
    private boolean doWaterPoolLevel(IotDataVO iotDataVO, EquipmentSpecificIndex equipmentSpecificIndex, Map<String, String> messageBody) {
        boolean alarmFlag = false;
        MessageModel model = new MessageModel();
        Map<String, Object> map = fireFightingSystemMapper.getWaterPoolLevel(equipmentSpecificIndex.getEquipmentSpecificId());
        if (!ObjectUtils.isEmpty(map)) {
            BigDecimal minValue = getBigDecimal(map.get("minValues"));
            BigDecimal maxValue = getBigDecimal(map.get("maxValues"));
            BigDecimal nowValue = getBigDecimal(iotDataVO.getValue());
            String levelDeviceId = (String) map.get("levelDeviceId");
            BigDecimal add = nowValue;
            if (levelDeviceId.contains(",")) {
                String[] split = levelDeviceId.split(",");
                int i = 1;
                for (String s : split) {
                    if (!s.trim().equals(Long.toString(equipmentSpecificIndex.getEquipmentSpecificId()))) {
                        BigDecimal waterPoolLevelByEquipSpeId = fireFightingSystemMapper.getWaterPoolLevelByEquipSpeId(s);
                        add = add.add(waterPoolLevelByEquipSpeId);
                        i++;
                    }
                }
                nowValue = add.divide(new BigDecimal(i), 2, RoundingMode.HALF_UP);
            }
            BigDecimal checkValue = checkUnit(equipmentSpecificIndex.getEquipmentSpecificId(), iotDataVO.getKey().toLowerCase(), nowValue);
            if (!ObjectUtils.isEmpty(checkValue)) {
                nowValue = checkValue;
            }
            if (nowValue.compareTo(minValue) < 0 || nowValue.compareTo(maxValue) > 0) {
                String body = "";
                if (nowValue.compareTo(minValue) < 0) {
                    body = "当前数值 " + nowValue + " 低于最低报警阈值 " + minValue;
                } else if (nowValue.compareTo(maxValue) > 0) {
                    body = "当前数值 " + nowValue + " 超过最高报警阈值 " + maxValue;
                }
                String bodyMain = String.format("%s，- 当前数值%s，%s ，请及时查看处理。",
                        map.get("name"),
                        nowValue,
                        nowValue.compareTo(minValue) < 0 ? "低于最低报警阈值" + minValue : "超过最高报警阈值" + maxValue
                );
                model.setTitle("模拟量提醒");
                model.setBody(bodyMain);
                model.setMsgType("WaterPoolKey");
                model.setSendTime(new Date());
                model.setIsSendWeb(true);
                model.setCategory(1);
                model.setRelationId(map.get("id").toString());
                model.setIsSendApp(false);
                model.setTerminal("WEB");
                model.setRecivers(Collections.singletonList("system"));
                Map<String, String> ext = new HashMap<>();
                ext.put("content", body);
                ext.put("type", "模拟量超阈值提醒");
                ext.put("name", (String) map.get("name"));
                ext.put("time", new SimpleDateFormat(DateUtils.DATE_TIME_PATTERN).format(new Date()));
                model.setExtras(ext);
//                Token token = remoteSecurityService.getServerToken();
//                systemctlFeign.create(token.getAppKey(), token.getProduct(), token.getToke(), model);
//                log.info(String.format("调用平台消息服务成功：%s", JSON.toJSONString(model)));
                alarmFlag = true;
                messageBody.put("messageBody", map.get("name") + "-" + body);
            }
        }
        return alarmFlag;
    }

    private BigDecimal checkUnit(Long equipmentSpecificId, String indexKey, BigDecimal nowValue) {
        LambdaQueryWrapper<EquipmentSpecificIndex> lambda = new QueryWrapper<EquipmentSpecificIndex>().lambda();
        lambda.eq(EquipmentSpecificIndex::getEquipmentSpecificId, equipmentSpecificId);
        lambda.eq(EquipmentSpecificIndex::getEquipmentIndexKey, indexKey);
        EquipmentSpecificIndex equipmentSpecificIndex = equipmentSpecificIndexMapper.selectOne(lambda);
        if (!ObjectUtils.isEmpty(equipmentSpecificIndex) || !ObjectUtils.isEmpty(equipmentSpecificIndex.getUnit())) {
            if (UnitEnum.MM.getKey().equalsIgnoreCase(equipmentSpecificIndex.getUnit()) || UnitEnum.MM.getName().equals(equipmentSpecificIndex.getUnit())) {
                BigDecimal divide = new BigDecimal(1000);
                nowValue = nowValue.divide(divide, 2, RoundingMode.HALF_UP);
            } else if (UnitEnum.CM.getKey().equalsIgnoreCase(equipmentSpecificIndex.getUnit()) || UnitEnum.CM.getName().equals(equipmentSpecificIndex.getUnit())) {
                BigDecimal divide = new BigDecimal(100);
                nowValue = nowValue.divide(divide, 2, RoundingMode.HALF_UP);
            }
        }
        return nowValue;
    }

    private BigDecimal getBigDecimal(Object value) {
        BigDecimal val = null;
        if (value != null) {
            if (value instanceof BigDecimal) {
                val = (BigDecimal) value;
            } else if (value instanceof String) {
                val = new BigDecimal((String) value);
            } else if (value instanceof BigInteger) {
                val = new BigDecimal((BigInteger) value);
            } else if (value instanceof Number) {
                val = new BigDecimal(String.valueOf(value));
            } else {
                throw new ClassCastException("Not possible to coerce [" + value + "] from class " + value.getClass() + " into a BigDecimal.");
            }
        }
        return val;
    }

    public void carRealTimeDate(List<IotDataVO> iotDatalist, List<CarProperty> carProperties, TopicEntityVo topicEntity) {
        List<CarProperty> carIndexsList = new ArrayList<>();
        iotDatalist.forEach(iotDataVO -> {
            // 对指标key为labels的数据处理
            if (EquipmentIndexLabelsEnum.labels.name.equals(iotDataVO.getKey())) {
                StringBuilder sb = new StringBuilder("equipmentOnCar_");
                EquipmentIndexLabelsVo labelsVo = new EquipmentIndexLabelsVo();
                Object obj = iotDataVO.getValue();
                if (obj instanceof JSONArray) {
                    List<String> labelList = (List<String>) obj;
                    labelList.forEach(code -> {
                        String key = sb.append(code).toString();
                        labelsVo.setEquipmentIotCode(code);
                        labelsVo.setTime(new Date());
                        redisUtils.set(key, com.alibaba.fastjson.JSONObject.toJSONString(labelsVo), redisExpireTime);
                    });
                }
            }
            List<CarPropertyVo> carPropertyVos = new ArrayList<>();
            carProperties.forEach(carProperty -> {
                if (iotDataVO.getKey().equals(carProperty.getNameKey())) {
                    carProperty.setValue(iotDataVO.getValue().toString());
                    carProperty.setUpdateDate(new Date());
                    carPropertyVos.add(carPropertyToCarPropertyVo(carProperty));
                    carIndexsList.add(carProperty);
                }
            });
            boolean updateBatchById = false;
            if (!carIndexsList.isEmpty()) {
                carPropertyService.updateBatchById(carIndexsList);
                updateBatchById = true;
            }
            if (updateBatchById) {
                carTransactionSynch(carProperties, carPropertyVos);
            }
        });

        //通知>消防应急预案

        mqttSendGateway.sendToMqtt(emergencyDisposalIndicators, JSONObject.toJSONString(topicEntity));

        // 需要在事务提交之后，否侧事务隔离查询不出数据
        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
            @Override
            public void afterCommit() {
                // 组态大屏消息推送，设备表实时指标修改
//                intePageSysDataRefresh(equipmentSpecificIndexList, topicEntity);
                mqttSendGateway.sendToMqtt(TopicEnum.EQXXTJ.getTopic(), String.valueOf(System.currentTimeMillis()));
            }
        });
    }

    /**
     * 报警消息推送amos平台
     *
     * @param equipmentSpecificAlarmLog
     */
    void syncSystemctlMsg(EquipmentSpecificAlarmLog equipmentSpecificAlarmLog) {
        try {
            MessageModel model = new MessageModel();
            String alarmReason = ValidationUtil.isEmpty(equipmentSpecificAlarmLog.getAlarmReason()) ? ""
                    : equipmentSpecificAlarmLog.getAlarmReason();
            if (alarmReason.contains("：")) {
                String[] split = alarmReason.split("：");
                alarmReason = split[1];
            }
            model.setTitle(equipmentSpecificAlarmLog.getEquipmentSpecificIndexName());
            String body = String.format("警情类型：%s；报警设备：%s；报警位置：%s；报警原因：%s；报警时间：%s",
                    equipmentSpecificAlarmLog.getEquipmentSpecificIndexName(),
                    equipmentSpecificAlarmLog.getEquipmentSpecificName(), equipmentSpecificAlarmLog.getLocation(),
                    alarmReason, DateUtils.date2LongStr(equipmentSpecificAlarmLog.getCreateDate()));
            model.setBody(body);
            model.setMsgType("iotMonitor");
            model.setSendTime(equipmentSpecificAlarmLog.getCreateDate());
            if (isSendApp) {
                model.setIsSendApp(true);
                model.setTerminal("APP/WEB");
            } else {
                model.setIsSendApp(false);
                model.setTerminal("WEB");
            }
            model.setIsSendWeb(true);
            model.setCategory(1);

            //告警弹窗需要根据人员单位判断是否展示 后端返回单位Code 前端判断
            LambdaQueryWrapper<EquipmentSpecific> wrapper = new LambdaQueryWrapper<>();
            wrapper.eq(BaseEntity::getId, equipmentSpecificAlarmLog.getEquipmentSpecificId());
            EquipmentSpecific equipmentSpecific = equipmentSpecificMapper.selectOne(wrapper);
            if (!ValidationUtil.isEmpty(equipmentSpecific) && !ValidationUtil.isEmpty(equipmentSpecific.getBizOrgCode())) {
                Map<String, String> equipMap = new HashMap<>();
                equipMap.put("bizOrgCode", equipmentSpecific.getBizOrgCode());
                model.setExtras(equipMap);
            }
            model.setRelationId(String.valueOf(equipmentSpecificAlarmLog.getId()));
            List<String> receive = new ArrayList<>();
            receive.add("system");
            model.setRecivers(receive);
            Token token = remoteSecurityService.getServerToken();
            systemctlFeign.create(token.getAppKey(), token.getProduct(), token.getToke(), model);
            log.info(String.format("调用平台消息服务成功：%s", JSON.toJSONString(model)));
        } catch (Exception e) {
            // e.printStackTrace();
            log.error("告警消息同步平台失败：syncSystemctlMsg，===>>>" + e.getMessage());
        }
    }

    private String getSystemCodeBySpeId(String sysIds) {
        if (StringUtil.isNotEmpty(sysIds)) {
            List<FireFightingSystemEntity> sys = fireFightingSystemMapper.getFightingSysByIds(sysIds.split(","));
            return sys.stream().map(FireFightingSystemEntity::getCode).collect(Collectors.joining(","));
        } else {
            return null;
        }
    }

    private EquipmentAlarmReportDay addEquipAlarmReportRecord(EquipmentSpecificIndex equipmentSpecificIndex, boolean flag) {
        EquipmentAlarmReportDay equipmentAlarmReportDay = new EquipmentAlarmReportDay();
        equipmentAlarmReportDay.setOrgCode(equipmentSpecificIndex.getOrgCode());
        equipmentAlarmReportDay.setReportDate(new Date());
        equipmentAlarmReportDay.setUpdateDate(new Date());
        equipmentAlarmReportDay.setSystemIds(equipmentSpecificIndex.getSystemId());
        equipmentAlarmReportDay.setLastReportDate(new Date());
        equipmentAlarmReportDay.setEquipmentSpecificId(equipmentSpecificIndex.getEquipmentSpecificId());
        equipmentAlarmReportDay.setEquipmentSpecificName(equipmentSpecificIndex.getEquipmentSpecificName());
        equipmentAlarmReportDay.setEquipmentDetailId(equipmentSpecificIndex.getEquipmentDetailId());
        equipmentAlarmReportDay.setEquipmentId(equipmentSpecificIndex.getEquipmentId());
        equipmentAlarmReportDay.setEquipmentCode(equipmentSpecificIndex.getEquipmentCode());
        if(flag) {
            equipmentAlarmReportDay
                    .setIndexTrueNum(1L);
        } else {
            equipmentAlarmReportDay
                    .setIndexTrueNum(TrueOrFalseEnum.real.value.equals(equipmentSpecificIndex.getValue()) ? 1L : 0L);
        }

        equipmentAlarmReportDay.setAlarmType(equipmentSpecificIndex.getTypeCode());
        equipmentAlarmReportDay.setAlarmTypeName(equipmentSpecificIndex.getTypeName());
        equipmentAlarmReportDay.setIndexId(equipmentSpecificIndex.getEquipmentIndexId());
        equipmentAlarmReportDay.setIndexName(equipmentSpecificIndex.getEquipmentSpecificIndexName());
        equipmentAlarmReportDay.setIndexType(equipmentSpecificIndex.getNameKey());
        equipmentAlarmReportDay.setValue(equipmentSpecificIndex.getValue());
        equipmentAlarmReportDay.setIsAlarm(equipmentSpecificIndex.getIsAlarm());
        return equipmentAlarmReportDay;
    }

    private EquipmentSpecificAlarmLog addEquipAlarmLogRecord(EquipmentSpecificAlarm equipmentSpecificAlarm) {
        EquipmentSpecificAlarmLog equipmentSpecificAlarmLog = new EquipmentSpecificAlarmLog();
        BeanUtils.copyProperties(equipmentSpecificAlarm, equipmentSpecificAlarmLog);
        equipmentSpecificAlarmLog.setId(null);
        equipmentSpecificAlarmLog.setCreateDate(new Date());
        equipmentSpecificAlarmLog.setEquipmentSpecificAlarmId(equipmentSpecificAlarm.getId());
        equipmentSpecificAlarmLog.setEquipmentSpecificIndexName(StringUtil.isNotEmpty(equipmentSpecificAlarm.getMessageBody()) ? equipmentSpecificAlarm.getMessageBody() : equipmentSpecificAlarm.getEquipmentSpecificIndexName());

        //调整为按照设备归属（可归属公司或者部门)
        Map<String, Object> mapd = iEquipmentSpecificSerivce.getStationCode(equipmentSpecificAlarm.getEquipmentSpecificId());
        if (mapd != null) {
            equipmentSpecificAlarmLog.setStationCode(mapd.get("station_code").toString());
            equipmentSpecificAlarmLog.setStationName(mapd.get("station_name").toString());
        }
//        equipmentSpecificAlarmLog.setStationCode(stationCode);
//        equipmentSpecificAlarmLog.setStationName(stationName);
        boolean bool = equipmentSpecificAlarmLogService.save(equipmentSpecificAlarmLog);
        // 同步告警消息给平台
        if (amosSwitch && bool) {
            EquipmentSpecificAlarmLog alarmLog = equipmentSpecificAlarmLogService
                    .getById(equipmentSpecificAlarmLog.getId());
            new Thread(new Runnable() {
                @Override
                public void run() {
                    syncSystemctlMsg(alarmLog);
                }
            }).start();
        }
        return equipmentSpecificAlarmLog;
    }

    /**
     * 添加告警
     *
     * @param equipmentSpecificAlarms
     * @param equipmentSpecificIndex
     * @return
     */
    private List<EquipmentSpecificAlarm> addEquipmentSpecificAlarm(List<EquipmentSpecificAlarm> equipmentSpecificAlarms,
                                                                   EquipmentSpecificIndex equipmentSpecificIndex, EquipmentSpecificAlarm equipmentSpecificAlarm, Map<String, String> messageBody) {
        BeanUtils.copyProperties(equipmentSpecificIndex, equipmentSpecificAlarm);
        equipmentSpecificAlarm.setId(null);
        equipmentSpecificAlarm.setFrequency(1);
        equipmentSpecificAlarm.setEquipmentSpecificName(equipmentSpecificIndex.getEquipmentSpecificName());
        equipmentSpecificAlarm.setStatus(AlarmStatusEnum.BJ.getCode());
        equipmentSpecificAlarm.setEquipmentSpecificIndexKey(equipmentSpecificIndex.getNameKey());
        equipmentSpecificAlarm.setEquipmentSpecificIndexValue(equipmentSpecificIndex.getValue());
        equipmentSpecificAlarm.setEquipmentSpecificIndexLabel(equipmentSpecificIndex.getValueLabel());
        equipmentSpecificAlarm.setType(equipmentSpecificIndex.getTypeCode());
        equipmentSpecificAlarm.setCreateDate(new Date());
        equipmentSpecificAlarm.setUpdateDate(new Date());
        equipmentSpecificAlarm.setEquipmentCode(equipmentSpecificIndex.getEquipmentCode());
        equipmentSpecificAlarm.setEquipmentId(equipmentSpecificIndex.getEquipmentId());
        equipmentSpecificAlarm.setEquipmentDetailId(equipmentSpecificIndex.getEquipmentDetailId());
        equipmentSpecificAlarm.setEquipmentSpecificCode(equipmentSpecificIndex.getEquipmentSpecificCode());
        equipmentSpecificAlarm.setEmergencyLevel(equipmentSpecificIndex.getEmergencyLevel());
        equipmentSpecificAlarm.setEmergencyLevelColor(equipmentSpecificIndex.getEmergencyLevelColor());
        equipmentSpecificAlarm.setEmergencyLevelDescribe(equipmentSpecificIndex.getEmergencyLevelDescribe());
        equipmentSpecificAlarm.setBizOrgCode(equipmentSpecificIndex.getBizOrgCode());
        equipmentSpecificAlarm.setBizOrgName(equipmentSpecificIndex.getBizOrgName());
        equipmentSpecificAlarm.setMessageBody(!ObjectUtils.isEmpty(messageBody) ? messageBody.get("messageBody") : "");

        //调整为按照设备归属（可归属公司或者部门
        Map<String, Object> mapd = iEquipmentSpecificSerivce.getStationCode(equipmentSpecificIndex.getEquipmentSpecificId());
        if (mapd != null) {
            equipmentSpecificAlarm.setStationCode(mapd.get("station_code").toString());
            equipmentSpecificAlarm.setStationName(mapd.get("station_name").toString());
        }

        equipmentSpecificAlarms.add(equipmentSpecificAlarm);
        return equipmentSpecificAlarms;
    }

    // NB装备告警
    private List<EquipmentSpecificAlarm> getNbEquipAlarmList(List<EquipmentSpecificAlarm> indexAlarms,
                                                             EquipmentSpecificIndex equipmentSpecificIndex, EquipmentSpecificAlarm equipmentSpecificAlarm, Map<String, String> messageBody) {
        List<EquipmentSpecificAlarm> equipmentSpecificAlarmList = new ArrayList<>();
        if (ValidationUtil.isEmpty(indexAlarms)) { // 告警表为空，新增告警数据
            addEquipmentSpecificAlarm(equipmentSpecificAlarmList, equipmentSpecificIndex, equipmentSpecificAlarm, messageBody);
            if (!checkStateIsNormal(equipmentSpecificAlarm, equipmentSpecificIndex)) {
                return equipmentSpecificAlarmList;
            } else {
                equipmentSpecificAlarmList.clear();
            }
        } else {
            indexAlarms.forEach(action -> {
                // 状态为正常或报警解除
                if (checkStateIsNormal(action, equipmentSpecificIndex)) {
                    // 修改报警数据为正常
                    action.setRecoveryDate(new Date());
                    action.setStatus(AlarmStatusEnum.HF.getCode());
                    // 修改之前数据为已处理
                    action.setResolveResult(action.getAlamReason());
                    action.setConfirmUserName("系统");
                    action.setConfirmType(action.getType());
                    equipmentSpecificAlarmLogService.updateAlarmLogByIotCodeAndIndexKey(action);
                } else {
                    action.setFrequency((action.getFrequency() + 1));
                }
                // 更新所在系统,设备可能编辑过
                action.setTraceId(equipmentSpecificIndex.getTraceId());
                action.setSystemIds(equipmentSpecificIndex.getSystemId());
                action.setSystemCodes(this.getSystemCodeBySpeId(equipmentSpecificIndex.getSystemId()));
                action.setUpdateDate(new Date());
                equipmentSpecificAlarmList.add(action);
            });
        }
        return equipmentSpecificAlarmList;
    }

    private boolean ifSendToGis(List<CarIndexGisVo> list) {
        boolean flag = true;
        List<CarIndexGisVo> list1 = list.stream().filter(x -> CarForGisEnum.JD.getNameKey().equals(x.getNameKey()))
                .collect(Collectors.toList());
        List<CarIndexGisVo> list2 = list.stream().filter(x -> CarForGisEnum.WD.getNameKey().equals(x.getNameKey()))
                .collect(Collectors.toList());
        for (CarIndexGisVo gisVo : list) {
            if (CarForGisEnum.JD.getNameKey().equals(gisVo.getNameKey())) {
                if (!StringUtil.isNotEmpty(gisVo.getValue()) || "0".equals(gisVo.getValue())) {
                    flag = false;
                    continue;
                }
            }
            if (CarForGisEnum.WD.getNameKey().equals(gisVo.getNameKey()) || "0".equals(gisVo.getValue())) {
                if (!StringUtil.isNotEmpty(gisVo.getValue())) {
                    flag = false;
                    continue;
                }
            }
        }
        if (list1.size() == 0 || list2.size() == 0) {
            flag = false;
        }
        return flag;
    }

    /**
     * 接收到的IOT数据为火眼存储到Map中
     */
    private void iotDataListToCacheMap(List<IotDataVO> iotDatalist) {
        List<IotDataVO> iotDataVOs = iotDatalist.stream()
                .filter(x -> "alarmLevel".equals(x.getKey()) || "alarmType".equals(x.getKey())
                        || "temperature".equals(x.getKey()) || "ruleTemperature".equals(x.getKey())
                        || "thermometryUnit".equals(x.getKey()) || "alarmRule".equals(x.getKey()))
                .collect(Collectors.toList());

//        List<IotDataVO> iotDataVOS = iotDatalist.stream()
//                .filter(x -> Stream.of("alarmLevel", "temperature", "thermometryUnit", "ruleTemperature", "alarmType", "alarmRule")
//                        .anyMatch(s -> s.equals(x.getKey()))).collect(Collectors.toList());

        if (iotDataVOs.size() > 0) {
            Map<String, Object> map = iotDatalist.stream()
                    .collect(Collectors.toMap(IotDataVO::getKey, IotDataVO::getValue));
            putTemperatureMap(map.get("traceId").toString(), map);
        }
    }

    /**
     * 处理火眼逻辑合并为一条告警
     */
    private EquipmentSpecificIndex handleTemperatureAlarm(EquipmentSpecificIndex equipmentSpecificIndex,
                                                          List<IotDataVO> iotDatalist) {
        List<IotDataVO> collect = iotDatalist.stream().filter(x -> "traceId".equals(x.getKey()))
                .collect(Collectors.toList());
        if (collect.size() > 0 && temperatureMapIsEmpty(String.valueOf(collect.get(0).getValue()))) {
            String traceId = String.valueOf(collect.get(0).getValue());
            TemperatureAlarmDto dto = temperatureMap.get(traceId);
            equipmentSpecificIndex
                    .setEquipmentSpecificIndexName(AlarmTypeEnum.getTypeByCode(AlarmTypeEnum.GZGJ.getCode()));
            equipmentSpecificIndex.setNameKey(AlarmTypeEnum.GZGJ.getCode());
            equipmentSpecificIndex.setAlamReason(
                    TemperatureAlarm.getAlarmContent(dto.getAlarmLevel(), dto.getAlarmType(), dto.getAlarmRule(),
                            dto.getRuleTemperature(), dto.getTemperature(), dto.getThermometryUnit()));
            equipmentSpecificIndex.setValue("true");
            equipmentSpecificIndex.setIsAlarm(1);
            temperatureMap.remove(traceId);
        }
        return equipmentSpecificIndex;
    }

    /**
     * temperatureMap存储数据
     */
    private void putTemperatureMap(String traceId, Map<String, Object> map) {
        TemperatureAlarmDto cacheTemperatureAlarmDto = temperatureMap.get(traceId);
        if (ValidationUtil.isEmpty(cacheTemperatureAlarmDto)) {
            cacheTemperatureAlarmDto = new TemperatureAlarmDto();
        }
        TemperatureAlarmDto newMap = JSON.parseObject(JSON.toJSONString(map), TemperatureAlarmDto.class);
        BeanUtil.copyPropertiesIgnoreNull(newMap, cacheTemperatureAlarmDto);
        temperatureMap.put(traceId, cacheTemperatureAlarmDto);
    }

    private boolean temperatureMapIsEmpty(String traceId) {
        TemperatureAlarmDto dto = temperatureMap.get(traceId);
        return !ValidationUtil.isEmpty(dto) && !ValidationUtil.isEmpty(dto.getAlarmLevel())
                && !ValidationUtil.isEmpty(dto.getAlarmType()) && !ValidationUtil.isEmpty(dto.getAlarmRule())
                && !ValidationUtil.isEmpty(dto.getRuleTemperature()) && !ValidationUtil.isEmpty(dto.getTemperature())
                && !ValidationUtil.isEmpty(dto.getThermometryUnit());
    }

    /**
     * 判断是否为NB物联监测设备参数
     *
     * @param nameKey
     * @return
     */
    private boolean verifyNB(String nameKey) {
        return nameKey.startsWith("NB_");
    }

    /**
     * 判断状态为正常或报警解除
     */
    private boolean checkStateIsNormal(EquipmentSpecificAlarm equipmentSpecificAlarm,
                                       EquipmentSpecificIndex equipmentSpecificIndex) {
        String enumKey = String.format("%s_%s", equipmentSpecificAlarm.getEquipmentSpecificIndexKey(),
                equipmentSpecificIndex.getValue());
        equipmentSpecificAlarm.setAlamReason(equipmentSpecificAlarm.getEquipmentSpecificIndexName().replace("NB_", "")
                + "：" + NBalarmEnum.getDescByKey(enumKey));
        boolean flag = !ValidationUtil.isEmpty(equipmentSpecificAlarm.getType())
                && !ValidationUtil.isEmpty(equipmentSpecificAlarm.getIotCode())
                && !ValidationUtil.isEmpty(equipmentSpecificAlarm.getEquipmentSpecificIndexKey())
                && (NBalarmEnum.NB_liquid_level_state_0.getKey().equals(enumKey)
                || NBalarmEnum.NB_error_code_0.getKey().equals(enumKey)
                || NBalarmEnum.NB_battery_state_0.getKey().equals(enumKey)
                || NBalarmEnum.NB_hydraulic_state_0.getKey().equals(enumKey)
                || NBalarmEnum.NB_hydraulic_state_2.getKey().equals(enumKey)
                || NBalarmEnum.NB_hydraulic_state_4.getKey().equals(enumKey)
                || NBalarmEnum.NB_alarm_status_4.getKey().equals(enumKey));
        return flag;
    }

//    public void saveOrUpdateEquipAlarm(
//            List<EquipmentSpecificAlarm> equipmentSpecificAlarms) {
//        equipmentSpecificAlarmService.saveOrUpdateBatch(equipmentSpecificAlarms);
//    }

    /**
     * 报警日志同步其他系统
     *
     * @param equipmentAlarmLogs
     */
    public void equipmentAlarmLogsToOtherSystems(List<EquipmentSpecificAlarmLog> equipmentAlarmLogs) throws InterruptedException {
        if (ObjectUtils.isEmpty(equipmentAlarmLogs)) {
            return;
        }
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("seqNo", UUID.randomUUID().toString().replace("-", "").toLowerCase());
        mqttSendGateway.sendToMqtt(TopicEnum.ALARM_LOG_INSERT.getTopic(), jsonObject.toString());
        // 数字换流站数据处理（高斯库同步及南瑞告警推送）
        if (syncSwitch) {
            List<FireEquipmentFireAlarm> alarmList = createFireEquipmentFireAlarmVo(equipmentAlarmLogs);
            if (!CollectionUtils.isEmpty(alarmList)) {
                Map<String, List<FireEquipmentFireAlarm>> collect = alarmList.stream()
                        .collect(Collectors.groupingBy(FireEquipmentFireAlarm::getType));
                for (String key : collect.keySet()) {
                    List<FireEquipmentFireAlarm> list = collect.get(key);
                    if (!CollectionUtils.isEmpty(list)) {
                        if ("FIREALARM".equalsIgnoreCase(key)) {
                            syncDataService.syncCreatedFireEquipAlarm(list);
                        } else if ("BREAKDOWN".equalsIgnoreCase(key)) {
                            List<FireEquipmentFaultAlarm> faultAlarms = list.stream().map(x -> {
                                FireEquipmentFaultAlarm fireEquipmentFaultAlarm = new FireEquipmentFaultAlarm();
                                BeanUtils.copyProperties(x, fireEquipmentFaultAlarm);
                                return fireEquipmentFaultAlarm;
                            }).collect(Collectors.toList());
                            syncDataService.syncCreatedFireEquiptFaultAlarm(faultAlarms);
                        } else if ("SHIELD".equalsIgnoreCase(key)) {
                            List<FireEquipmentDefectAlarm> defectAlarms = list.stream().map(x -> {
                                FireEquipmentDefectAlarm fireEquipmentFaultAlarm = new FireEquipmentDefectAlarm();
                                BeanUtils.copyProperties(x, fireEquipmentFaultAlarm);
                                return fireEquipmentFaultAlarm;
                            }).collect(Collectors.toList());
                            syncDataService.syncCreatedFireEquipDefectAlarm(defectAlarms);
                        }
                    }
                }
            }
        }
        // 向南瑞平台推送报警消息
        log.info("判断条件bool值为：" + JSON.toJSONString(bool) + ", NRMsgSend值为：" + JSON.toJSONString(NRMsgSend));
        if (!bool && NRMsgSend) {
            syncDataService.syncCreatedSendAlarm(equipmentAlarmLogs);
        }
    }

    /**
     * 组装数字换流站平台告警数据
     *
     * @param
     * @return
     */
    private List<FireEquipmentFireAlarm> createFireEquipmentFireAlarmVo(
            List<EquipmentSpecificAlarmLog> equipmentAlarmLogs) {
        Map<String, String> stationInfo = equipmentSpecificMapper.getStationInfo().get(0);
        List<FireEquipmentFireAlarm> alarmList = new ArrayList<>();
        equipmentAlarmLogs.forEach(action -> {
            FireEquipmentFireAlarm alarm = new FireEquipmentFireAlarm();
            BeanUtils.copyProperties(action, alarm);
            alarm.setAliasname(StringUtil.toNotEmptyString(action.getEquipmentSpecificIndexName()));
            alarm.setEquipmentMeasurementId(StringUtil.toNotEmptyString(action.getEquipmentIndexId().toString()));
            alarm.setEquipmentMeasurementMRid(StringUtil.toNotEmptyString(action.getEquipmentIndexId().toString()));
            alarm.setFieldLabel(StringUtil.toNotEmptyString(action.getEquipmentSpecificIndexKey()));
            alarm.setFieldName(StringUtil.toNotEmptyString(action.getEquipmentSpecificIndexName()));
            alarm.setFireEquipmentId(StringUtil.toNotEmptyString(action.getEquipmentSpecificId().toString()));
            alarm.setFireEquipmentMRid(StringUtil.toNotEmptyString(action.getEquipmentSpecificCode()));
            alarm.setFireEquipmentName(StringUtil.toNotEmptyString(action.getEquipmentSpecificName()));
            alarm.setFrequency(1);
            alarm.setId(StringUtil.toNotEmptyString(action.getId().toString()));
            alarm.setMrid(action.getId().toString());
            alarm.setName(action.getEquipmentSpecificIndexName());
            alarm.setRecoveryDate(action.getUpdateDate());
            alarm.setStationCode(StringUtil.toNotEmptyString(stationInfo.get("stationCode")));
            alarm.setStationName(StringUtil.toNotEmptyString(stationInfo.get("stationName")));
            alarm.setValue(StringUtil.toNotEmptyString(action.getEquipmentSpecificIndexValue()));
            alarmList.add(alarm);
        });
        return alarmList;
    }

    /**
     * 高斯库同步指标修改
     *
     * @param equipmentSpecificIndexList
     */
    private void syncSpecificIndexsToGS(List<EquipmentSpecificIndex> equipmentSpecificIndexList) {
        if (!ObjectUtils.isEmpty(equipmentSpecificIndexList) && syncSwitch) {
            // 数据同步
            List<EquipmentIndexVO> fireEquipMeasurementCollect = new ArrayList<>();
            equipmentSpecificIndexList.forEach(action -> {
                EquipmentIndexVO equipmentIndexVO = new EquipmentIndexVO();
                BeanUtils.copyProperties(action, equipmentIndexVO);
                fireEquipMeasurementCollect.add(equipmentIndexVO);
            });
            if (0 < fireEquipMeasurementCollect.size()) {
                syncDataService.syncCreatedFireEquipMeasurement(fireEquipMeasurementCollect);
            }
        }
    }

    private IndexStateVo createIndexStateVo(EquipmentSpecificIndex equipmentSpecificIndex) {
        IndexStateVo indexStateVo = new IndexStateVo();
        BeanUtils.copyProperties(equipmentSpecificIndex, indexStateVo);
        indexStateVo.setId(equipmentSpecificIndex.getIotCode() + "_" + equipmentSpecificIndex.getNameKey());
        indexStateVo.setData(equipmentSpecificIndex.getValue());
        indexStateVo.setIndexKey(equipmentSpecificIndex.getNameKey());
        return indexStateVo;
    }

    public List<EquipmentSpecificAlarm> createIndexAlarmRecord(EquipmentSpecificIndex equipmentSpcIndex, Map<String, String> messageBody) {
        // 处理火眼视频异常
        List<EquipmentSpecificAlarm> equipmentSpecificAlarms = new ArrayList<>();
        EquipmentSpecificAlarm equipmentSpecificAlarm = new EquipmentSpecificAlarm();
        equipmentSpecificAlarm.setSystemIds(equipmentSpcIndex.getSystemId());
        equipmentSpecificAlarm.setSystemCodes(this.getSystemCodeBySpeId(equipmentSpcIndex.getSystemId()));
        List<EquipmentSpecificAlarm> indexAlarms = equipmentSpecificAlarmMapper
                .findEquipmentSpecificAlarmByEquipmentSpecificIdAndEquipmentIndexIdAndStatusIstrue(
                        equipmentSpcIndex.getEquipmentSpecificId(), equipmentSpcIndex.getEquipmentIndexId());
        // NB设备告警
        if (verifyNB(equipmentSpcIndex.getNameKey())) {
            return getNbEquipAlarmList(indexAlarms, equipmentSpcIndex, equipmentSpecificAlarm, messageBody);
        }

        Boolean flag = judgeIsAlarm(equipmentSpcIndex.getAlarmRule(), equipmentSpcIndex.getValue());
        if (StringUtil.isNotEmpty(equipmentSpcIndex.getAlarmRule())) {
            if (ObjectUtils.isEmpty(indexAlarms) && flag) {
                equipmentSpecificAlarms = addEquipmentSpecificAlarm(equipmentSpecificAlarms, equipmentSpcIndex, equipmentSpecificAlarm, messageBody);
            } else {
                for (EquipmentSpecificAlarm action : indexAlarms) {
                    if (flag) {
                        // 报警，修改发生频率
                        action.setFrequency((action.getFrequency() + 1));
                    } else {
                        // 报警恢复，修改数据为恢复状态
                        action.setRecoveryDate(new Date());
                        action.setEquipmentSpecificIndexValue(equipmentSpcIndex.getValue());
                        action.setStatus(AlarmStatusEnum.HF.getCode());
                    }
                    action.setUpdateDate(new Date());
                    // 更新所在系统,设备可能编辑过,更新所在系统、装备名称、装备定义code
                    action.setTraceId(equipmentSpcIndex.getTraceId());
                    action.setSystemIds(equipmentSpcIndex.getSystemId());
                    action.setSystemCodes(this.getSystemCodeBySpeId(equipmentSpcIndex.getSystemId()));
                    action.setEquipmentSpecificName(equipmentSpcIndex.getEquipmentSpecificName());
                    action.setEquipmentCode(equipmentSpcIndex.getEquipmentCode());
                    // 冗余字段，alarm_log表更新时使用
                    action.setEquipmentSpecificCode(equipmentSpcIndex.getEquipmentSpecificCode());
                    action.setBuildId(equipmentSpcIndex.getBuildId());
                    action.setBizOrgName(equipmentSpcIndex.getBizOrgName());
                    action.setBizOrgCode(equipmentSpcIndex.getBizOrgCode());
                    action.setMessageBody(!ObjectUtils.isEmpty(messageBody) ? messageBody.get("messageBody") : "");
                    equipmentSpecificAlarms.add(action);
                }
            }
        } else {
            // 报警表新增信息
            if (ObjectUtils.isEmpty(indexAlarms) && (TrueOrFalseEnum.real.value.equals(equipmentSpcIndex.getValue()))) {
                equipmentSpecificAlarms = addEquipmentSpecificAlarm(equipmentSpecificAlarms, equipmentSpcIndex, equipmentSpecificAlarm, messageBody);
            } else {
                // 报警表更新信息
                for (EquipmentSpecificAlarm action : indexAlarms) {
                    if (TrueOrFalseEnum.real.value.equals(equipmentSpcIndex.getValue())) {
                        // 报警，修改发生频率
                        action.setFrequency((action.getFrequency() + 1));
                    } else {
                        // 报警恢复，修改数据为恢复状态
                        action.setRecoveryDate(new Date());
                        action.setEquipmentSpecificIndexValue(equipmentSpcIndex.getValue());
                        action.setStatus(AlarmStatusEnum.HF.getCode());
                    }
                    action.setUpdateDate(new Date());
                    // 更新所在系统,设备可能编辑过,更新所在系统、装备名称、装备定义code
                    action.setTraceId(equipmentSpcIndex.getTraceId());
                    action.setSystemIds(equipmentSpcIndex.getSystemId());
                    action.setSystemCodes(this.getSystemCodeBySpeId(equipmentSpcIndex.getSystemId()));
                    action.setEquipmentSpecificName(equipmentSpcIndex.getEquipmentSpecificName());
                    action.setEquipmentCode(equipmentSpcIndex.getEquipmentCode());
                    // 冗余字段，alarm_log表更新时使用
                    action.setEquipmentSpecificCode(equipmentSpcIndex.getEquipmentSpecificCode());
                    action.setBuildId(equipmentSpcIndex.getBuildId());
                    action.setBizOrgName(equipmentSpcIndex.getBizOrgName());
                    action.setBizOrgCode(equipmentSpcIndex.getBizOrgCode());
                    action.setMessageBody(!ObjectUtils.isEmpty(messageBody) ? messageBody.get("messageBody") : "");
                    equipmentSpecificAlarms.add(action);
                }
            }
        }
        return equipmentSpecificAlarms;
    }

    /**
     * 根据alarm_rule规则字段判断是否是告警
     * @return
     */
    private Boolean judgeIsAlarm(String rule, String value) {
        Boolean bool = Boolean.TRUE;
        if (StringUtil.isNotEmpty(rule)) {
            if (!rule.equalsIgnoreCase(value)) {
                bool = Boolean.FALSE;
            }
        }
        return bool;
    }

    /**
     * 发送数据至换流站
     *
     * @param
     */
    public void sendEquipSpecIndexToAutosysTopic(List<EquipmentSpecificIndex> equipmentSpeIndexs) {
        equipmentSpeIndexs.forEach(equipmentSpeIndex -> {
            String topic = "";
            if (TrueOrFalseEnum.real.value.equals(equipmentSpeIndex.getValue())
                    && EquipmentRiskTypeEnum.HZGJ.getCode().equals(equipmentSpeIndex.getTypeCode())) {
                equipmentSpeIndex.setType(EquipmentRiskTypeEnum.HZGJ.getCode());
                topic = String.format("%s.%s%s", serverName, "equipment/", RiskLeverForAutoSys.BJ.getCode());
            } else if (TrueOrFalseEnum.real.value.equals(equipmentSpeIndex.getValue())
                    && EquipmentRiskTypeEnum.GZ.getCode().equals(equipmentSpeIndex.getTypeCode())) {
                equipmentSpeIndex.setType(EquipmentRiskTypeEnum.GZ.getCode());
                topic = String.format("%s.%s%s", serverName, "equipment/", RiskLeverForAutoSys.GZ.getCode());
            } else {
                equipmentSpeIndex.setType(EquipmentRiskTypeEnum.QT.getCode());
                topic = String.format("%s.%s%s", serverName, "equipment/", RiskLeverForAutoSys.JC.getCode());
            }
            TopicEntityVo topicEntityVo = new TopicEntityVo();
            topicEntityVo.setIotCode(equipmentSpeIndex.getIotCode());
            topicEntityVo.setTopic(topic);
            topicEntityVo.setMessage(JSON.toJSONString(equipmentSpeIndex));
            mqttSendGateway.sendToMqtt(topic, JSON.toJSONString(topicEntityVo));
        });
    }

    /**
     * 组态大屏消息推送
     *
     * @param equipmentSpecificIndexList
     * @param topicEntity
     */
    public void intePageSysDataRefresh(List<EquipmentSpecificIndex> equipmentSpecificIndexList,
                                       TopicEntityVo topicEntity) {
        mqttSendGateway.sendToMqtt(TopicEnum.EQXXTJ.getTopic(), String.valueOf(System.currentTimeMillis()));
        iEquipmentSpecificSerivce.integrationPageSysDataRefresh(topicEntity.getCode());
    }

    public void publishDataToDCCenterPage(List<EquipmentSpecificIndex> equipmentSpecificIndexList) {
        mqttSendGateway.sendToMqtt(TopicEnum.EQZXDT.getTopic(), JSON.toJSONString(equipmentSpecificIndexList));
        log.info(String.format("直流中心指标消息发送：%s", JSON.toJSONString(equipmentSpecificIndexList)));
    }

    public void publishNormalIndexValueToPage(List<EquipmentSpecificIndex> equipmentSpecificIndexList) {
        mqttSendGateway.sendToMqtt(TopicEnum.SHBZYCXH.getTopic(), JSON.toJSONString(equipmentSpecificIndexList));
        log.info(String.format("四横八纵遥测指标消息发送：%s", JSON.toJSONString(equipmentSpecificIndexList)));
    }

    public void publishDataToMessage(List<EquipmentSpecificIndex> equipmentSpeIndexes, Boolean isAlarm) {
        equipmentSpeIndexes.forEach(equipmentSpeIndex -> {
            SendToMsgRiskDetailVo detailVo = new SendToMsgRiskDetailVo();
            SendToMsgRiskEquipInfoVo equipInfoVo = new SendToMsgRiskEquipInfoVo();
            SendToMsgRiskAlarmInfoVo alarmInfoVo = new SendToMsgRiskAlarmInfoVo();
            equipInfoVo.setId(String.valueOf(equipmentSpeIndex.getEquipmentSpecificId()));
            equipInfoVo.setName(equipmentSpeIndex.getEquipmentSpecificName());
            equipInfoVo.setCode(equipmentSpeIndex.getEquipmentSpecificCode());
            equipInfoVo.setBizOrgCode(equipmentSpeIndex.getBizOrgCode());
            equipInfoVo.setBizOrgName(equipmentSpeIndex.getBizOrgName());
            equipInfoVo.setReportIsAlarm(isAlarm);
            equipInfoVo.setEquipSpeId(String.valueOf(equipmentSpeIndex.getEquipmentSpecificId()));

            alarmInfoVo.setKey(equipmentSpeIndex.getEquipmentIndexKey());
            alarmInfoVo.setValue(equipmentSpeIndex.getValue());
            alarmInfoVo.setAlarmContent(equipmentSpeIndex.getEquipmentName() + equipmentSpeIndex.getEquipmentIndexName());
            alarmInfoVo.setAlarmDate(equipmentSpeIndex.getUpdateDate());

            TopicEntityVo topicEntityVo = new TopicEntityVo();
            topicEntityVo.setIotCode(equipmentSpeIndex.getIotCode());
            topicEntityVo.setTopic(riskTopicCenter);
            detailVo.setEquipInfo(equipInfoVo);
            detailVo.setAlarmInfo(alarmInfoVo);
            topicEntityVo.setData(JSON.toJSONString(detailVo));
            mqttSendGateway.sendToMqtt(riskTopicCreate, JSON.toJSONString(topicEntityVo));
            log.info(String.format("站端发送给中心级触发风险：%s", JSON.toJSONString(topicEntityVo)));
        });
    }

    @Override
    public void handleDataToRiskModel(String topic, String message) {
        JSONObject jsonObject = JSONObject.parseObject(message);
        SendToMsgRiskDetailVo detailVo = JSONObject.toJavaObject(jsonObject, SendToMsgRiskDetailVo.class);
        SendToMsgRiskEquipInfoVo equipInfoVo = detailVo.getEquipInfo();
        SendToMsgRiskAlarmInfoVo equipAlarm = detailVo.getAlarmInfo();
        if (StringUtil.isNotEmpty(detailVo.getEquipInfo().getEquipSpeId())) {
            EquipmentSpecific specific = equipmentSpecificMapper.selectById(detailVo.getEquipInfo().getEquipSpeId());
            equipInfoVo.setPosition(specific.getPosition());
            equipInfoVo.setManufacturerName(getEquipmentDetailInfo(specific.getEquipmentDetailId()).getManufacturerName());
            equipInfoVo.setSystemIds(specific.getSystemId());
            equipInfoVo.setSystemName(getSystemNameBySpeId(specific));
            equipInfoVo.setQrCode(specific.getQrCode());
            equipInfoVo.setCodeColor(EquipQrcodeColorEnum.getEnumByKey(specific.getEquipStatus()).getColor());
            equipInfoVo.setBatchNo(getEquipmentDetailInfo(specific.getEquipmentDetailId()).getStandard());
        }

        RiskDataVo dataVo = new RiskDataVo();
        dataVo.setIndexKey(detailVo.getAlarmInfo().getKey());
        dataVo.setIndexValue(detailVo.getAlarmInfo().getValue());
        RiskBizInfoVo infoVo = new RiskBizInfoVo();
        infoVo.setWarningObjectName(detailVo.getEquipInfo().getName());
        infoVo.setWarningObjectCode(detailVo.getEquipInfo().getCode());
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        infoVo.setWarningTime(sdf.format(new Date()));
        infoVo.setSourceAttribution(detailVo.getEquipInfo().getBizOrgCode());
        infoVo.setSourceAttributionDesc(detailVo.getEquipInfo().getBizOrgName());
        infoVo.setWarningObjectCode(detailVo.getEquipInfo().getId());
        List<RiskDynamicDetailsVo> detailsVos = new ArrayList<>();
        RiskDynamicDetailsVo dynamicDetailsVo = new RiskDynamicDetailsVo();
        dynamicDetailsVo.setTabName("装备指标联动数据");
        List<TableContentVo> contentVos = new ArrayList<>();
        for (DynamicLabelEnum u : DynamicLabelEnum.values()) {
            TableContentVo contentVo = new TableContentVo();
            contentVo.setKey(u.getCode());
            switch (u.getCode()) {
                case "name":
                    contentVo.setLabel(u.getDescribe());
                    contentVo.setValue(equipInfoVo.getName());
                    contentVo.setType("text");
                    break;
                case "code":
                    contentVo.setLabel(u.getDescribe());
                    contentVo.setValue(equipInfoVo.getCode());
                    contentVo.setType("text");
                    break;
                case "position":
                    contentVo.setLabel(u.getDescribe());
                    contentVo.setValue(equipInfoVo.getPosition());
                    contentVo.setType("text");
                    break;
                case "manufacturer":
                    contentVo.setLabel(u.getDescribe());
                    contentVo.setValue(equipInfoVo.getManufacturerName());
                    contentVo.setType("text");
                    break;
                case "systemName":
                    contentVo.setLabel(u.getDescribe());
                    contentVo.setValue(equipInfoVo.getSystemName());
                    contentVo.setType("text");
                    break;
                case "qrCode":
                    contentVo.setLabel(u.getDescribe());
                    Map<String, Object> map = new HashMap<>();
                    map.put("qrCode", equipInfoVo.getQrCode());
                    map.put("color", equipInfoVo.getCodeColor());
                    contentVo.setValue(map);
                    contentVo.setType("qrcode");
                    break;
                case "alarmContent":
                    contentVo.setLabel(u.getDescribe());
                    contentVo.setValue(equipAlarm.getAlarmContent());
                    contentVo.setType("text");
                    break;
                case "alarmDate":
                    contentVo.setLabel(u.getDescribe());
                    contentVo.setValue(equipAlarm.getAlarmDate());
                    contentVo.setType("text");
                    break;
                default:
                    log.info("方法参数错误 !!!");
                    break;
            }
            contentVos.add(contentVo);
        }

        dynamicDetailsVo.setTabContent(contentVos);
        detailsVos.add(dynamicDetailsVo);
        infoVo.setDynamicDetails(detailsVos);
        dataVo.setBizInfo(infoVo);
        dataVo.setTraceId(detailVo.getEquipInfo().getTraceId());
        mqttSendGateway.sendToMqtt(riskTopic, JSON.toJSONString(dataVo));
    }

    public String getSystemNameBySpeId(EquipmentSpecific equipmentSpecific) {
        List<FireFightingSystemEntity> sys = new ArrayList<>();
        String[] ids;
        String sysIds = equipmentSpecific.getSystemId();
        if (StringUtil.isNotEmpty(sysIds)) {
            if (-1 != sysIds.indexOf(",")) {
                ids = sysIds.split(",");
                sys = fireFightingSystemMapper.getFightingSysByIds(ids);
            } else {
                FireFightingSystemEntity entity = fireFightingSystemMapper.selectById(Long.valueOf(sysIds));
                sys.add(entity);
            }
        }
        StringBuilder sb = new StringBuilder();
        sys.forEach(x -> {
            if (0 < sb.length()) {
                sb.append(",");
            }
            sb.append(x.getName());
        });
        return sb.toString();
    }

    public EquipmentDetail getEquipmentDetailInfo(long id) {
        EquipmentDetail detail = equipmentDetailMapper.selectById(id);
        if (ObjectUtils.isEmpty(detail.getManufacturerId())) {
            detail.setManufacturerName("");
            return detail;
        }
        ManufacturerInfo info = manufacturerInfoMapper.selectById(detail.getManufacturerId());
        detail.setManufacturerName(info.getName());
        return Optional.ofNullable(detail).orElse(detail);
    }

    /**
     * 更新数据报表表
     *
     * @param equipmentSpecificIndex
     */
    private void saveEquipmentAlarmReportDay(EquipmentSpecificIndex equipmentSpecificIndex, boolean flag) {
        SimpleDateFormat sdf = new SimpleDateFormat(DateUtils.DATE_PATTERN);
        EquipmentAlarmReportDay equipmentAlarmReportDay = addEquipAlarmReportRecord(equipmentSpecificIndex, flag);
        LambdaQueryWrapper<EquipmentAlarmReportDay> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(EquipmentAlarmReportDay::getReportDate, sdf.format(equipmentAlarmReportDay.getReportDate()))
                .eq(EquipmentAlarmReportDay::getIndexId, equipmentAlarmReportDay.getIndexId())
                .eq(EquipmentAlarmReportDay::getEquipmentSpecificId, equipmentAlarmReportDay.getEquipmentSpecificId());
        List<EquipmentAlarmReportDay> reportDayList = iEquipmentAlarmReportDayService.list(wrapper);
        if (reportDayList.isEmpty()) {
            equipmentAlarmReportDay.setReportDate(new Date());
            equipmentAlarmReportDay.setFrequency(1);
            iEquipmentAlarmReportDayService.save(equipmentAlarmReportDay);
        } else {
            EquipmentAlarmReportDay reportDay = reportDayList.get(0);
            reportDay.setLastReportDate(new Date());
            reportDay.setValue(equipmentAlarmReportDay.getValue());
            reportDay.setFrequency(reportDay.getFrequency() + 1);
            reportDay.setIndexTrueNum(reportDay.getIndexTrueNum() == null ? equipmentAlarmReportDay.getIndexTrueNum()
                    : reportDay.getIndexTrueNum() + equipmentAlarmReportDay.getIndexTrueNum());
            iEquipmentAlarmReportDayService.updateById(reportDay);
        }

    }

    private String valueTranslate(String value, String enumStr) {
        if (ObjectUtils.isEmpty(enumStr)) {
            return "";
        }
        try {
            JSONArray jsonArray = JSONArray.parseArray(enumStr);
            for (int i = 0; i < jsonArray.size(); i++) {
                JSONObject jsonObject = jsonArray.getJSONObject(i);
                if (jsonObject.get("key").equals(value)) {
                    return jsonObject.getString("label");
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "";
    }

    /**
     * 车辆数据推送及同步
     *
     * @param carProperties
     * @param carPropertyVos
     */
    public void carTransactionSynch(List<CarProperty> carProperties, List<CarPropertyVo> carPropertyVos) {
        // TODO 数字化换流站组态屏数据推送,需要在事务提交之后，否侧事务隔离查询不出数据
        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
            @Override
            public void afterCommit() {
                List<CarIndexGisVo> list = createCarIndexGisVo(carProperties);
                mqttSendGateway.sendToMqtt(TopicEnum.CARZXDT.getTopic(), "");
                boolean flag = ifSendToGis(list);
                if (flag) {
                    mqttSendGateway.sendToMqtt(carTopic, JSON.toJSONString(list));
                }

                if (syncSwitch) {
                    syncDataService.syncCreatedFireVehicleMeasurement(carPropertyVos);
                }
            }
        });
    }

    private CarPropertyVo carPropertyToCarPropertyVo(CarProperty property) {
        CarPropertyVo carPropertyVo = new CarPropertyVo();
        carPropertyVo.setCarId(property.getCarId());
        carPropertyVo.setCreateDate(property.getCreateDate());
        carPropertyVo.setId(property.getId());
        carPropertyVo.setIsIot(1);
        carPropertyVo.setMRid(property.getEquipmentIndexId().toString());
        carPropertyVo.setName(property.getEquipmentIndexName());
        carPropertyVo.setNameKey(property.getEquipmentIndexKey());
        carPropertyVo.setSort(1);
        carPropertyVo.setUnit(property.getUnitName());
        carPropertyVo.setValue(property.getValue());
        return carPropertyVo;
    }

    private List<CarIndexGisVo> createCarIndexGisVo(List<CarProperty> carProperties) {
        List<CarIndexGisVo> list = new ArrayList<>();
        long id = 0L;
        String iotCode = "";
        for (CarProperty action : carProperties) {
            CarIndexGisVo v = new CarIndexGisVo();
            id = action.getCarId();
            iotCode = action.getIotCode();
            v.setId(action.getCarId());
            v.setIotCode(action.getIotCode());
            v.setNameKey(action.getEquipmentIndexKey());
            v.setValue(ObjectUtils.isEmpty(action.getValue()) ? "0" : action.getValue());
            list.add(v);
        }

        CarIndexGisVo time = new CarIndexGisVo();
        time.setId(id);
        time.setIotCode(iotCode);
        time.setNameKey(CarForGisEnum.SJ.getNameKey());
        time.setValue(String.valueOf(new Date().getTime()));
        list.add(time);

        return list;
    }

    /**
     * //若为物联设备，则更新拓扑节点数据及告警状态
     *
     * @param indexList
     */
    public void updateNodeDateByEquipId(List<EquipmentSpecificIndex> indexList) {
        if (!ObjectUtils.isEmpty(indexList)) {
            EquipmentVo equipmentVo = equipmentService.getEquipBySpecific(indexList.get(0).getEquipmentSpecificId());
            if ("1".equals(equipmentVo.getIsIot())) {
                List<EquipmentSpecificAlarm> alarmList = equipmentSpecificAlarmService.getEquipListBySpecific(true,
                        indexList.get(0).getEquipmentSpecificId());
                topographyService.updateNodeDateByEquipId(indexList.get(0).getEquipmentSpecificId(), indexList,
                        alarmList);
            }
        }
    }

//    private void pressurePump(String indexKey, String indexValue, List<IotDataVO> iotDatalist, TopicEntityVo topicEntity) {
//        List<String> listIndex = new ArrayList<>();
//        listIndex.add(pressurePumpStart);
//        // 获取全部启停泵信号
//        List<EquipmentSpecificIndex> equipmentSpeIndexList = equipmentSpecificIndexService.getEquipmentSpeIndexByIndex(listIndex);
//        List<PressurePumpEnum> enumListByCode = PressurePumpEnum.getEnumListByCode(indexValue);
//
//        if (!CollectionUtils.isEmpty(enumListByCode)) {
//            enumListByCode.forEach(pressurePumpEnum -> {
//                // 1. 获取需要校验的值
//                PressurePumpValueEnum valueEnum = PressurePumpValueEnum.getByCode(pressurePumpEnum.getCompareValue());
//                assert valueEnum != null;
//                EquipmentSpecificIndex data = getPressurePumpDateByType(indexKey, valueEnum, topicEntity, equipmentSpeIndexList, pressurePumpEnum);
//                Date newDate = new Date();
//                // 2. 校验
//                if (!ObjectUtils.isEmpty(data.getUpdateDate())) {
//                    checkValueByDate(data, newDate, pressurePumpEnum);
//                } else {
//                    // 稳压泵漏水告警恢复
//                    List<EquipmentSpecificIndex> collect = equipmentSpeIndexList.stream().filter(item -> !ObjectUtils.isEmpty(item.getIotCode()) && item.getIotCode().equals(topicEntity.getIotCode())).collect(Collectors.toList());
//                    if (!ObjectUtils.isEmpty(collect) && !ObjectUtils.isEmpty(collect.get(0)) && !ObjectUtils.isEmpty(collect.get(0).getEquipmentId())) {
//                        equipmentSpecificAlarmLogService.pressurePumpRestore(collect.get(0).getEquipmentId());
//                    }
//                }
//            });
//        }
//    }

    private void pressurePump(String indexKey, String indexValue, List<IotDataVO> iotDatalist, TopicEntityVo topicEntity) {
        List<String> listIndex = new ArrayList<>();
        listIndex.add(pressurePumpStart);
        // 获取全部启停泵信号
        List<EquipmentSpecificIndex> equipmentSpeIndexList = equipmentSpecificIndexService.getEquipmentSpeIndexByIndex(listIndex);
        List<PressurePumpEnum> enumListByCode = PressurePumpEnum.getEnumListByCode(indexValue);

        if (TrueOrFalseEnum.fake.value.equals(indexValue)) {
            String jobName = topicEntity.getIotCode() + "_" + indexKey;
            String triggerName = PUMP_TRIGGER_NAME + "-" + topicEntity.getIotCode();
            boolean b = QuartzManager.checkExists(jobName, PUMP_JOB_GROUP_NAME);
            // 删除这个稳压泵的监听任务
            if (b) {
                QuartzManager.removeJob(jobName, PUMP_JOB_GROUP_NAME, triggerName, PUMP_TRIGGER_GROUP_NAME);
            }

            // 稳压泵漏水告警恢复
            List<EquipmentSpecificIndex> collect = equipmentSpeIndexList.stream().filter(item -> !ObjectUtils.isEmpty(item.getIotCode()) && item.getIotCode().equals(topicEntity.getIotCode())).collect(Collectors.toList());
            if (!ObjectUtils.isEmpty(collect) && !ObjectUtils.isEmpty(collect.get(0)) && !ObjectUtils.isEmpty(collect.get(0).getEquipmentId())) {
                equipmentSpecificAlarmLogService.pressurePumpRestore(collect.get(0).getEquipmentId());
            }
        }
        if (!CollectionUtils.isEmpty(enumListByCode)) {
            enumListByCode.forEach(pressurePumpEnum -> {
                // 1. 获取需要校验的值
                PressurePumpValueEnum valueEnum = PressurePumpValueEnum.getByCode(pressurePumpEnum.getCompareValue());
                assert valueEnum != null;
                EquipmentSpecificIndex data = getPressurePumpDateByType(indexKey, valueEnum, topicEntity, equipmentSpeIndexList, pressurePumpEnum);
                Date newDate = new Date();
                // 2. 校验
                if (!ObjectUtils.isEmpty(data.getUpdateDate())) {
                    checkValueByDate(data, newDate, pressurePumpEnum);
                } else {
                    // 稳压泵漏水告警恢复
                    List<EquipmentSpecificIndex> collect = equipmentSpeIndexList.stream().filter(item -> !ObjectUtils.isEmpty(item.getIotCode()) && item.getIotCode().equals(topicEntity.getIotCode())).collect(Collectors.toList());
                    if (!ObjectUtils.isEmpty(collect) && !ObjectUtils.isEmpty(collect.get(0)) && !ObjectUtils.isEmpty(collect.get(0).getEquipmentId())) {
                        equipmentSpecificAlarmLogService.pressurePumpRestore(collect.get(0).getEquipmentId());
                    }
                }
            });
        }
    }

    private EquipmentSpecificIndex getPressurePumpDateByType(String indexKey, PressurePumpValueEnum valueEnum, TopicEntityVo topicEntity, List<EquipmentSpecificIndex> equipmentSpeIndexList, PressurePumpEnum pressurePumpEnum) {
        String iotCode = topicEntity.getIotCode();
        EquipmentSpecificIndex equipmentSpecificIndex = new EquipmentSpecificIndex();
        String prefix = null;
        String suffix = null;
        if (iotCode.length() > 8) {
            prefix = iotCode.substring(0, 8);
            suffix = iotCode.substring(8);
        } else {
            throw new BadRequest("装备物联编码错误，请确认！");
        }

        switch (valueEnum) {
            case LAST_STOP:
                List<EquipmentSpecificIndex> lastStop = equipmentSpeIndexList.stream().filter(e ->
                                StringUtil.isNotEmpty(e.getValue()) && e.getIotCode().equals(iotCode) && pressurePumpStart.equals(e.getEquipmentIndexKey())).sorted(Comparator.comparing(EquipmentSpecificIndex::getUpdateDate).reversed())
                        .collect(Collectors.toList());
                if (!CollectionUtils.isEmpty(lastStop)) {
                    EquipmentSpecificIndex aFalse = getIotDate(equipmentSpecificIndex, lastStop, prefix, suffix, "false");
                    BeanUtils.copyProperties(aFalse, equipmentSpecificIndex);
                }
                break;
            case LAST_START:
                List<EquipmentSpecificIndex> lastStart = equipmentSpeIndexList.stream().filter(e ->
                                StringUtil.isNotEmpty(e.getValue()) && e.getIotCode().equals(iotCode) && pressurePumpStart.equals(e.getEquipmentIndexKey())).sorted(Comparator.comparing(EquipmentSpecificIndex::getUpdateDate).reversed())
                        .collect(Collectors.toList());
                if (!CollectionUtils.isEmpty(lastStart)) {
                    EquipmentSpecificIndex aTrue = getIotDate(equipmentSpecificIndex, lastStart, prefix, suffix, "true");
                    BeanUtils.copyProperties(aTrue, equipmentSpecificIndex);
                }
                break;
            case LATELY_STOP:
                List<EquipmentSpecificIndex> latelyStop = equipmentSpeIndexList.stream().filter(e ->
                                StringUtil.isNotEmpty(e.getValue()) && pressurePumpStart.equals(e.getEquipmentIndexKey())).sorted(Comparator.comparing(EquipmentSpecificIndex::getUpdateDate).reversed())
                        .collect(Collectors.toList());
                if (!CollectionUtils.isEmpty(latelyStop)) {
                    EquipmentSpecificIndex aFalse = getIotDate(equipmentSpecificIndex, latelyStop, prefix, null, "false");
                    BeanUtils.copyProperties(aFalse, equipmentSpecificIndex);

                }
                break;
            case LATELY_START:
                List<EquipmentSpecificIndex> latelyStart = equipmentSpeIndexList.stream().filter(e ->
                                StringUtil.isNotEmpty(e.getValue()) && e.getIotCode().equals(iotCode) && pressurePumpStart.equals(e.getEquipmentIndexKey())).sorted(Comparator.comparing(EquipmentSpecificIndex::getUpdateDate).reversed())
                        .collect(Collectors.toList());
                if (!CollectionUtils.isEmpty(latelyStart)) {
                    EquipmentSpecificIndex aTrue = getIotDateExceptSelf(equipmentSpecificIndex, latelyStart, prefix, suffix, "true");
                    BeanUtils.copyProperties(aTrue, equipmentSpecificIndex);
                }
                break;
            case PUMP_START_TIME:
                startTimeCompute(indexKey, topicEntity, pressurePumpEnum);
                break;
            default:
                break;
        }
        return equipmentSpecificIndex;
    }


    private EquipmentSpecificIndex getIotDate(EquipmentSpecificIndex equipmentSpecificIndex, List<EquipmentSpecificIndex> listData, String prefix, String suffix, String flag) {
        ResponseModel start = iotFeign.selectOne(remoteSecurityService.getServerToken().getAppKey(), remoteSecurityService.getServerToken().getProduct(), remoteSecurityService.getServerToken().getToke(), "2", prefix, suffix, flag, pressurePumpStart);
        if (200 == start.getStatus() && !ObjectUtils.isEmpty(start.getResult())) {
            String json1 = JSON.toJSONString(start.getResult());
            List<Map<String, String>> listObject1 = (List<Map<String, String>>) JSONArray.parse(json1);
            List<Map<String, String>> collect = listObject1.stream().filter(t -> (t.containsKey("time"))).collect(Collectors.toList());
            Date startDate = null;
            if (1 < collect.size()) {
                String startTime = collect.get(1).get("time").substring(0, 19).replace("T", " ");
                startDate = DateUtils.dateAddHours(DateUtils.longStr2Date(startTime), +8);
            }
            listData.get(0).setUpdateDate(startDate);
            BeanUtils.copyProperties(listData.get(0), equipmentSpecificIndex);
        }
        return equipmentSpecificIndex;
    }

    private EquipmentSpecificIndex getIotDateExceptSelf(EquipmentSpecificIndex equipmentSpecificIndex, List<EquipmentSpecificIndex> listData, String prefix, String suffix, String flag) {
        ResponseModel start = iotFeign.selectOne(remoteSecurityService.getServerToken().getAppKey(), remoteSecurityService.getServerToken().getProduct(), remoteSecurityService.getServerToken().getToke(), "100", prefix, null, flag, pressurePumpStart + "," + "deviceName");
        if (200 == start.getStatus() && !ObjectUtils.isEmpty(start.getResult())) {
            String json1 = JSON.toJSONString(start.getResult());
            List<Map<String, String>> listObject1 = (List<Map<String, String>>) JSONArray.parse(json1);
            listObject1 = listObject1.stream().filter(x -> !suffix.equalsIgnoreCase(x.get("deviceName"))).collect(Collectors.toList());
            List<Map<String, String>> collect = listObject1.stream().filter(t -> (t.containsKey("time"))).collect(Collectors.toList());
            Date startDate = null;
            if (0 < collect.size()) {
                String startTime = collect.get(0).get("time").substring(0, 19).replace("T", " ");
                startDate = DateUtils.dateAddHours(DateUtils.longStr2Date(startTime), +8);
            }
            listData.get(0).setUpdateDate(startDate);
            BeanUtils.copyProperties(listData.get(0), equipmentSpecificIndex);
        }
        return equipmentSpecificIndex;
    }

    private void checkValueByDate(EquipmentSpecificIndex data, Date newDate, PressurePumpEnum pressurePumpEnum) {
        String operator = pressurePumpEnum.getOperator();
        PressurePumpCheckEnum pumpCheckEnum = PressurePumpCheckEnum.getByCode(operator);
        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        long diff = 0;
        try {
            long d1 = df.parse(df.format(data.getUpdateDate())).getTime();
            long d2 = df.parse(df.format(newDate)).getTime();
            diff = (d2 - d1) / 1000 / 60;
        } catch (Exception e) {
            log.error("时间转换失败" + e.getMessage());
            return;
        }
        assert pumpCheckEnum != null;
        String leftValue = pressurePumpEnum.getLeftValue();
        String rightValue = pressurePumpEnum.getRightValue();
        switch (pumpCheckEnum) {
            case LE:
                if (StringUtil.isNotEmpty(rightValue)) {
                    long value = Long.parseLong(rightValue);
                    if (diff <= value) {
                        sendMessagePressure(pressurePumpEnum, data);
                    }
                }
                break;
            case GE:
                if (StringUtil.isNotEmpty(leftValue)) {
                    long value = Long.parseLong(leftValue);
                    if (diff >= value) {
                        sendMessagePressure(pressurePumpEnum, data);
                    }
                }
                break;
            case BE:
                if (StringUtil.isNotEmpty(leftValue) && StringUtil.isNotEmpty(rightValue)) {
                    long lvalue = Long.parseLong(leftValue);
                    long rvalue = Long.parseLong(rightValue);
                    if (diff >= lvalue && diff <= rvalue) {
                        sendMessagePressure(pressurePumpEnum, data);
                    }
                }
                break;
            default:
                break;
        }
    }

    private void sendMessagePressure(PressurePumpEnum pressurePumpEnum, EquipmentSpecificIndex data) {
        String level = pressurePumpEnum.getLevel();
        PressurePumpMessageEnum pumpMessageEnum = PressurePumpMessageEnum.getByCode(level);
        assert pumpMessageEnum != null;
        String time = new SimpleDateFormat(DateUtils.DATE_TIME_PATTERN).format(new Date());
        MessageModel model = new MessageModel();
        String allMessage = pumpMessageEnum.getAllMessage();
        String body = "";
        Map<String, String> map = new HashMap<>(4);
        if (StringUtil.isNotEmpty(allMessage)) {
            String value = StringUtil.isNotEmpty(pressurePumpEnum.getTips()) ? pressurePumpEnum.getTips() : "";
            String content = String.format(allMessage, value);
            map.put("content", content);
            map.put("name", data.getEquipmentSpecificName());
            map.put("time", time);
        }
        String recordMessage = pumpMessageEnum.getRecordMessage();
        if (StringUtil.isNotEmpty(recordMessage)) {
            String value = StringUtil.isNotEmpty(pressurePumpEnum.getTips()) ? pressurePumpEnum.getTips() : "";
            body = String.format(recordMessage, data.getEquipmentSpecificName(), data.getLocation(), value);
        }
        String marqueeMessage = pumpMessageEnum.getMarqueeMessage();
        if (StringUtil.isNotEmpty(marqueeMessage)) {
            MarqueeData marqueeData = new MarqueeData();
            marqueeData.setIsRead(0);
            marqueeData.setEquipmentSpecificId(data.getEquipmentId());
            marqueeData.setMessageType("pressurePump");
            marqueeData.setName(data.getEquipmentSpecificName());
            marqueeData.setPosition(data.getLocation());
            marqueeData.setType("漏水提醒");
            marqueeData.setCreateDate(new Date());
            marqueeDataMapper.insert(marqueeData);
            // 稳压泵告警事件生成
            equipmentSpecificAlarmLogService.pressurePumpDisposeAlarm(data.getEquipmentId(), String.format(org.apache.commons.lang3.StringUtils.substring(recordMessage, 13), pressurePumpEnum.getTips()));
        }

        switch (pumpMessageEnum) {
            case MESSAGE_LEVEL_YB:
            case MESSAGE_LEVEL_YZ:
                map.put("type", "稳压泵启停异常提醒");
                model.setTitle("稳压泵启停异常提醒");
                break;
            case MESSAGE_LEVEL_QT_WJ:
            case MESSAGE_LEVEL_QT_WJ_YXSC:
                map.put("type", "漏水提醒");
                model.setTitle("漏水提醒");
                break;
            default:
                break;
        }

        model.setSendTime(new Date());
        model.setBody(body);
        model.setExtras(map);
        model.setMsgType("pressurePump");
        model.setIsSendApp(false);
        model.setTerminal("WEB");
        model.setIsSendWeb(true);
        model.setCategory(1);
        List<String> receive = new ArrayList<>();
        receive.add("system");
        model.setRelationId(String.valueOf(data.getEquipmentId()));
        model.setRecivers(receive);
        Token token = remoteSecurityService.getServerToken();
        systemctlFeign.create(token.getAppKey(), token.getProduct(), token.getToke(), model);
    }

    private void startTimeCompute(String indexKey, TopicEntityVo topicEntity, PressurePumpEnum pressurePumpEnum) {
        String jobName = topicEntity.getIotCode() + "_" + indexKey;


        String cron = "";
        String triggerName = PUMP_TRIGGER_NAME + "-" + topicEntity.getIotCode();

        if ("FHS_PressurePump_Start_ALONE_START_YXSC".equals(pressurePumpEnum.getCode())) {

            Calendar time = Calendar.getInstance();
            time.add(Calendar.MINUTE, 5);
            cron = time.get(Calendar.SECOND) + " " + time.get(Calendar.MINUTE) + "/5 * * * ?";

        } else {
            cron = pressurePumpEnum.getLeftValue();
        }

        EquipmentSpecific equipmentSpecific = null;
        try {
            LambdaQueryWrapper<EquipmentSpecific> wrapper = new LambdaQueryWrapper<>();
            wrapper.eq(EquipmentSpecific::getIotCode, topicEntity.getIotCode());
            equipmentSpecific = equipmentSpecificMapper.selectOne(wrapper);
        } catch (Exception e) {
            log.error("根据iotCod查询失败" + topicEntity.getIotCode());
        }
        boolean b = QuartzManager.checkExists(jobName, PUMP_JOB_GROUP_NAME);
        if (indexKey.equals(pressurePumpStart)) {
            if (b) {
                // 任务存在  更新时间
                QuartzManager.modifyJobTime(triggerName, PUMP_TRIGGER_GROUP_NAME, cron);
            } else {
                QuartzManager.removeJob(jobName, PUMP_JOB_GROUP_NAME, triggerName, PUMP_TRIGGER_GROUP_NAME);
                // 任务不存在，新增
                // 传参
                if (ObjectUtils.isEmpty(equipmentSpecific)) {
                    return;
                }
                Map<String, Object> parameter = new HashMap<>(6);
                parameter.put("jobName", jobName);
                parameter.put("triggerName", triggerName);
                parameter.put("triggerGroupName", PUMP_TRIGGER_GROUP_NAME);
                parameter.put("jobGroupName", PUMP_JOB_GROUP_NAME);
                parameter.put("equipmentSpecific", equipmentSpecific);
                parameter.put("pressurePumpEnum", pressurePumpEnum);
                parameter.put("remoteSecurityService", remoteSecurityService);
                parameter.put("systemctlFeign", systemctlFeign);
                parameter.put("marqueeDataMapper", marqueeDataMapper);
                parameter.put("equipmentSpecificAlarmLogService", equipmentSpecificAlarmLogService);
                QuartzManager.addJob(jobName, PUMP_JOB_GROUP_NAME, triggerName, PUMP_TRIGGER_GROUP_NAME, PumpSendMessage.class, cron, parameter);
            }
        }
    }


    @Async
    public void pushDataToIntegrationPage(List<EquipmentSpecificIndex> specificIndices) {
        for (EquipmentSpecificIndex specificIndex : specificIndices) {
            Long equipmentSpecificId = specificIndex.getEquipmentSpecificId();
            List<EquipmentSpecificIndex> equipmentSpecificIndices = equipmentSpecificIndexMapper.selectList(
                    Wrappers.<EquipmentSpecificIndex>lambdaQuery()
                            .eq(EquipmentSpecificIndex::getEquipmentSpecificId, equipmentSpecificId)
                            .isNotNull(EquipmentSpecificIndex::getValue)
                            .ne(EquipmentSpecificIndex::getValue, "")
                            .orderByDesc(EquipmentSpecificIndex::getEmergencyLevel)
            );
            List<HashMap<String, String>> valuedIndexes = equipmentSpecificIndices.stream().map(index -> new HashMap<String, String>() {{
                put("key", index.getEquipmentIndexKey());
                put("value", index.getValue());
            }}).collect(Collectors.toList());
            JSONObject message1 = new JSONObject() {{
                put("code", String.valueOf(specificIndex.getId()));
                put("status", specificIndex.getValue());
                put("value", specificIndex.getValue());
            }};
            JSONObject message2 = new JSONObject() {{
                put("code", String.valueOf(specificIndex.getEquipmentSpecificId()));
                put("valuedIndexes", valuedIndexes);
            }};
            mqttSendGateway.sendToMqtt(ConfigPageTopicEnum.EQUIP_INDICATOR.getTopic(), message1.toJSONString());
            mqttSendGateway.sendToMqtt(ConfigPageTopicEnum.EQUIP_MULTI_INDICATOR.getTopic(), message2.toJSONString());
        }
    }
}
