NB_NetPump/coffee-system/src/main/java/com/coffee/bus/config/HospitalFunctionAnalConfigHandler.java

package com.coffee.bus.config;


import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.coffee.bus.config.bean.FunctionAnalConfig;
import com.coffee.bus.entity.BusDeviceRunningEntity;
import com.coffee.bus.entity.BusInfusionHistoryEntity;
import com.coffee.bus.registry.device.DeviceRegistry;
import com.coffee.bus.service.LocalBusDeviceRunningService;
import com.coffee.bus.service.LocalBusInfusionHistoryService;
import com.coffee.bus.utils.WsPublishUtils;
import com.coffee.common.cache.ClusterConfigStorage;
import com.coffee.common.util.RedissonUtil;
import lombok.Builder;
import lombok.Data;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RDelayedQueue;

import java.io.Serializable;
import java.util.Date;
import java.util.Optional;
import java.util.concurrent.TimeUnit;

/**
 * @author lifang
 * @version 1.0.0
 * @ClassName HospitalFunctionAnalConfigHandler.java
 * @Description 镇痛不足检测
 * @createTime 2022年05月18日 11:25:00
 */
@Slf4j
public class HospitalFunctionAnalConfigHandler extends ClusterConfigStorage implements HospitalConfigHandler<FunctionAnalConfig,BusDeviceRunningEntity> {

    private final LocalBusDeviceRunningService deviceRunningService;

    private final LocalBusInfusionHistoryService infusionHistoryService;

    private final RedissonUtil redissonUtil;

    private final DeviceRegistry deviceRegistry;
    @Getter
    private String name;

    public HospitalFunctionAnalConfigHandler(String name, LocalBusDeviceRunningService deviceRunningService, LocalBusInfusionHistoryService infusionHistoryService, RedissonUtil redissonUtil, DeviceRegistry deviceRegistry, WsPublishUtils wsPublishUtils) {
        super(redissonUtil.getRedissonClient(), name);
        this.name=name;
        this.deviceRunningService = deviceRunningService;
        this.infusionHistoryService = infusionHistoryService;
        this.redissonUtil = redissonUtil;
        this.deviceRegistry = deviceRegistry;
        this.wsPublishUtils = wsPublishUtils;
    }

    private WsPublishUtils wsPublishUtils;



    @Override
    public String getId() {
        return "insufficient";
    }

    @Override
    public String getDescription() {
        return "进行镇痛不足判定、不在服务区判定、低输注判定";
    }


    @Override
    public void setConfig(FunctionAnalConfig config) {
        this.setConfig("config",config);
    }

    @Override
    public void handler(BusDeviceRunningEntity source) {
        FunctionAnalConfig config = this.getConfig("config").as(FunctionAnalConfig.class);
        if(config==null){
            log.warn("id:{},配置名称:{}，不存在",getId(),name);
            return;
        }
        handlerAnal(source,config);
    }

    /**
     * 描述: 镇痛不足处理
     * 在时间窗口中判断pca是否达到一定次数
     * @author lifang
     * @date 2022/5/18 14:05
     * @param source
     * @param analConfig
     * @return void
     */
    private void handlerAnal(BusDeviceRunningEntity source,FunctionAnalConfig analConfig){
        RDelayedQueue delayedQueue = redissonUtil.getDelayedQueue(getId() + "-" + source.getTenantId() + "-" + source.getDeviceId(), e -> {
            if(e instanceof AnalEntity){
                this.judgeAnalPoor((AnalEntity) e);
            }
        });
        //如果有新的输注产生,则清空延迟队列
        if(source.isNewInfusion()){
            //镇痛不足设置为false
            source.setWarnAnalgesicPoor(false);
            delayedQueue.clear();
        }
        Integer insufficientTime = analConfig.getInsufficientTime();
        Integer insufficientCount = analConfig.getInsufficientCount();
        Boolean valid = analConfig.getValid();
        if(valid==null||(insufficientTime==null||insufficientTime<=0)||
                (insufficientCount==null||insufficientCount<=0)){
            return;
        }

        AnalEntity anal = AnalEntity.builder()
                .deviceId(source.getDeviceId())
                .pcaValidCount(source.getPcaValidCount())
                .pcaInvalidCount(source.getPcaInvalidCount())
                .timeout(analConfig.getInsufficientTime())
                .unit(TimeUnit.SECONDS)
                .threshold(analConfig.getInsufficientCount())
                .judgeByValid(analConfig.getValid())
                .tenantId(source.getTenantId())
                .timestamp(new Date())
                .build();

        delayedQueue.offer(anal,anal.getTimeout(),anal.getUnit());
    }


    /**
     * 描述: 判断延迟任务是否触发镇痛不足
     * @author lifang
     * @date 2022/5/19 9:14
     * @param
     * @return void
     */
    private void judgeAnalPoor(AnalEntity anal){
        //镇痛消失延迟队列
        RDelayedQueue noneAnalDelayedQueue = redissonUtil.getDelayedQueue("none-" + getId()+"-" + anal.getTenantId()+"-" + anal.getDeviceId(), e -> {
            if(e instanceof NoneAnalEntity){
                handleNoneAnal((NoneAnalEntity) e);
            }
        });
        String deviceId = anal.getDeviceId();
        BusDeviceRunningEntity runningInfo = deviceRunningService.getOne(new QueryWrapper<BusDeviceRunningEntity>().lambda().eq(BusDeviceRunningEntity::getDeviceId, deviceId).eq(BusDeviceRunningEntity::getTenantId, anal.getTenantId()));
        if(runningInfo==null){
            //泵已换绑医院，无需再处理
            return;
        }
        //PCA差值
        int subCount=0;
        if(anal.isJudgeByValid()){
            subCount=Math.subtractExact(runningInfo.getPcaValidCount(), Optional.ofNullable(anal.getPcaValidCount()).orElse(0));
        }else {
            subCount=Math.subtractExact(runningInfo.getPcaInvalidCount(), Optional.ofNullable(anal.getPcaInvalidCount()).orElse(0));
        }
        if(subCount>anal.getThreshold()){
            log.info("设备：{}镇痛不足",deviceId);
            //触发阈值,设置为镇痛不足
            if(!Boolean.TRUE.equals(runningInfo.getWarnAnalgesicPoor())){
                deviceRunningService.update(new UpdateWrapper<BusDeviceRunningEntity>().lambda().eq(BusDeviceRunningEntity::getId,runningInfo.getId())
                        .set(BusDeviceRunningEntity::getWarnAnalgesicPoor,true));
                infusionHistoryService.update(new UpdateWrapper<BusInfusionHistoryEntity>().lambda().eq(BusInfusionHistoryEntity::getId,runningInfo.getInfusionId())
                        .set(BusInfusionHistoryEntity::getWarnAnalgesicPoor,true));
                //报警/提醒缓存重置
                deviceRegistry.getOperator(deviceId).setAlarmOrWarn(null);
                if(Boolean.TRUE.equals(runningInfo.getMaster())){
                    //发布推送
                    wsPublishUtils.publishPatientMonitor(runningInfo.getPatientCode(),runningInfo.getTenantId());
                }
                //设置镇痛不足后,设置镇痛消失延迟时间
            }
            noneAnalDelayedQueue.clear();
        }else {
            //没有触发阈值,发入镇痛消失处理
            FunctionAnalConfig config = this.getConfig("config").as(FunctionAnalConfig.class);
            NoneAnalEntity noneAnal = NoneAnalEntity.builder()
                    .deviceId(anal.getDeviceId())
                    .tenantId(anal.getTenantId())
                    .timeout(config.getDisappearTime())
                    .unit(TimeUnit.SECONDS)
                    .timestamp(new Date())
                    .build();
            noneAnalDelayedQueue.offer(noneAnal,noneAnal.getTimeout(),noneAnal.getUnit());
        }
    }


    private void handleNoneAnal(NoneAnalEntity noneAnal){
        String deviceId = noneAnal.getDeviceId();
        BusDeviceRunningEntity runningInfo = deviceRunningService.getOne(new QueryWrapper<BusDeviceRunningEntity>().lambda().eq(BusDeviceRunningEntity::getDeviceId, deviceId).eq(BusDeviceRunningEntity::getTenantId, noneAnal.getTenantId()));
        if(runningInfo==null){
            //泵已换绑医院，无需再处理
            return;
        }
        //将设备由镇痛不足取消成功
        if (deviceRunningService.update(new UpdateWrapper<BusDeviceRunningEntity>().lambda().eq(BusDeviceRunningEntity::getId, runningInfo.getId())
                .eq(BusDeviceRunningEntity::getWarnAnalgesicPoor, true)
                .set(BusDeviceRunningEntity::getWarnAnalgesicPoor, false))) {
            log.info("设备：{}取消镇痛不足",deviceId);
            infusionHistoryService.update(new UpdateWrapper<BusInfusionHistoryEntity>().lambda().eq(BusInfusionHistoryEntity::getId,runningInfo.getInfusionId())
                    .set(BusInfusionHistoryEntity::getWarnAnalgesicPoor,false));
            deviceRegistry.getOperator(deviceId).setAlarmOrWarn(null);
            if(Boolean.TRUE.equals(runningInfo.getMaster())){
                //发布推送
                wsPublishUtils.publishPatientMonitor(runningInfo.getPatientCode(),runningInfo.getTenantId());
            }
        }



    }
    @Data
    @Builder
    static class AnalEntity implements Serializable {
        private String deviceId;
        private Integer pcaValidCount;
        private Integer pcaInvalidCount;
        private Integer timeout;
        private TimeUnit unit;
        private Integer threshold;
        private boolean judgeByValid;
        private String tenantId;
        private Date timestamp;
    }

    @Data
    @Builder
    static class NoneAnalEntity implements Serializable {
        private String deviceId;
        private Integer timeout;
        private TimeUnit unit;
        private String tenantId;
        private Date timestamp;
    }

}