twin-service/ruoyi-admin/src/main/java/com/ruoyi/biz/service/impl/AsyncServiceImpl.java

package com.ruoyi.biz.service.impl;

import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import com.ruoyi.biz.domain.*;
import com.ruoyi.biz.service.IIotService;
import com.ruoyi.biz.tools.Tools;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.*;
import java.util.concurrent.Future;
import java.util.stream.Collectors;

/**
 * 多线程执行任务
 *
 * @author wukai
 * @date 2024/5/4 20:35
 */
@Service
@Slf4j
public class AsyncServiceImpl {
    @Resource
    private IIotService iotService;

    @Async("threadPoolTaskExecutor")
    public Future<Map<String, Object>> currData(TwinDevice twinDevice) {
        String table = twinDevice.getDevicePath();
        String sql = "select last * from " + table;
        JSONObject jsonObject = iotService.query(sql);
        JSONObject data = jsonObject.getJSONObject("data");
        JSONArray values = data.getJSONArray("values");

        Map<String, Object> dataMap = Tools.json2Map(values, table);
        dataMap.put("device", twinDevice);
        dataMap.put("total", values.size());
        return new AsyncResult<>(dataMap);
    }

    @Async("threadPoolTaskExecutor")
    public Future<Map<String, List<?>>> process(TwinDevice twinDevice, Date date, long startTime, long endTime, int period) {
        Map<String, List<?>> result = new HashMap<>(16);
        List<TwinCalcHour> calcHours = new ArrayList<>();
        String table = twinDevice.getDevicePath();
        TwinCalcHour hour = new TwinCalcHour();
        hour.setDeviceId(twinDevice.getDeviceId());
        hour.setDataDate(date);
        hour.setHour(period);

        Map<String, Object> map = calc(table, startTime, endTime);


        //0.已织造米数  1.总能耗 2.总重量
        float[] total = (float[]) map.get("total");
        hour.setLength(BigDecimal.valueOf(total[0]));
        hour.setWeight(BigDecimal.valueOf(total[2]));
        hour.setKwh(BigDecimal.valueOf(total[1]));
        hour.setOpenTime((long) total[3]);
        hour.setCloseTime((long) total[4]);
        calcHours.add(hour);

        List<TwinPanHeadInfo> panHeadInfo = (List<TwinPanHeadInfo>) map.get("panHead");
        panHeadInfo.forEach(info -> info.setDeviceId(twinDevice.getDeviceId()));

        List<TwinCalcHourSpec> specs = (List<TwinCalcHourSpec>) map.get("specList");
        specs.forEach(spec -> {
            spec.setDeviceId(twinDevice.getDeviceId());
            spec.setDataDate(date);
            spec.setHour(period);
        });

        List<TwinRecordStop> stopRecord = (List<TwinRecordStop>) map.get("stopRecord");
        stopRecord.forEach(stop -> {
            stop.setDeviceId(twinDevice.getDeviceId());
            stop.setDataDate(date);
            stop.setHour(period);
        });
        List<TwinRecordAlarms> alarmRecord = (List<TwinRecordAlarms>) map.get("alarmRecord");
        alarmRecord.forEach(alarm -> {
            alarm.setDeviceId(twinDevice.getDeviceId());
            alarm.setDataDate(date);
            alarm.setHour(period);
        });

        result.put("calc", calcHours);
        result.put("stopRecord", stopRecord);
        result.put("alarmRecord", alarmRecord);
        result.put("panHead", panHeadInfo);
        result.put("specList", specs);
        return new AsyncResult<>(result);
    }

    /**
     * 字段列表，方便看下标
     */
    private final String[] fields = {"Capacity_data_2", "Capacity_data_37", "Capacity_data_38", "Capacity_data_39", "Capacity_data_42",
            "Capacity_data_43", "Capacity_data_44", "Capacity_data_48", "Formula_data_3", "Formula_data_13",
            "Capacity_data_36", "Capacity_data_41", "Alarm_unit_1", "Alarm_unit_2", "Alarm_unit_3",
            "Alarm_unit_4", "Alarm_unit_5", "Alarm_unit_6", "Alarm_unit_7", "Alarm_unit_8",
            "Alarm_unit_9", "Alarm_unit_10", "Alarm_unit_11", "Alarm_unit_12", "Alarm_unit_13",
            "Alarm_unit_14", "Alarm_unit_15", "Alarm_unit_16", "Alarm_unit_17", "Alarm_unit_18",
            "Alarm_unit_19", "Alarm_unit_20", "Alarm_unit_21", "Alarm_unit_22", "Alarm_unit_23",
            "Alarm_unit_24", "Alarm_unit_25", "Alarm_unit_26", "Alarm_unit_27", "Capacity_data_33",
            "Capacity_data_15", "Capacity_data_16", "Capacity_data_17", "Capacity_data_18", "Capacity_data_19",
            "Capacity_data_34", "Formula_data_24"
    };


    /**
     * 字段列表，方便查找位置
     */
    private final List<String> fieldList = Arrays.stream(fields).collect(Collectors.toList());


    /**
     * //    0        Capacity_data_2	已织造米数
     * //    1        Capacity_data_37	A班组开机时间
     * //    2        Capacity_data_38	A班组停机时间
     * //    3        Capacity_data_39	A班当前产量
     * //    4        Capacity_data_42	B班组开机时间
     * //    5        Capacity_data_43	B班组停机时间
     * //    6        Capacity_data_44	B班当前产量
     * //    7        Capacity_data_48	停机状态
     * //    8        Formula_data_3	米克重
     * //    9        Formula_data_13	卷曲幅宽
     */
    public Map<String, Object> calc(String table, long startTime, long endTime) {
        String sql = "select %s from %s where time>%s and time <=%s";
        sql = String.format(sql, Arrays.stream(fields).collect(Collectors.joining(",")), table, startTime, endTime);
        long s = System.currentTimeMillis();
        JSONObject jsonObject = iotService.query(sql);
        long e = System.currentTimeMillis();
        log.info("接口耗时:{}ms,table:{},time:{}", e - s, table, new Date(endTime));
        JSONObject data = jsonObject.getJSONObject("data");
        JSONArray values = data.getJSONArray("values");
        JSONArray timestamps = data.getJSONArray("timestamps");
        //初始时间点数据
        //0-data2,1-data37,2-data38,3-data39,4-data42,5-data-43,6data-44
        //2024-06-26 1-6不要了 0织造米长 1.电量
        List<TwinCalcHourSpec> specList = new ArrayList<>();
        //1.米长 2.电量  3.密度
        float[] first = new float[3];
        //上一个时间点数据
        float[] last = new float[3];
        //统计数据,后面2个分别代表0.织造米长,1.电量 2.重量,3.开机时间,4.停机时间
        float[] total = new float[5];
        //上一个时间点盘头数据
        int[] lastPanHead = new int[5];
        //米克重
        int lastMkz = 0;
        //卷曲幅宽
        float lastFk = 0f;
        //上次密度记录的米长
        float lastSpecLength = 0f;
        //开始不为0的电能
        //不为0的电能结束
        float startKwh = 0f, endKwh = 0f;
        int last48 = 0;
        //上一轮最后一条停机记录
        JSONArray old = null;
        //上一轮最后一条告警记录
//        Object oldV = CacheUtils.get(Constants.IOT_TOKEN, table + "-stop");
//        if (oldV != null) {
//            //获取存储的最后一次停机状态值
//            old = (JSONArray) oldV;
//        }
        //告警时间记录
        List<TwinRecordAlarms> alarmRecord = new ArrayList<>();
        //停机记录
        List<TwinRecordStop> stopRecord = new ArrayList<>();
        //盘头记录
        List<TwinPanHeadInfo> panHeadInfo = new ArrayList<>();

        for (int i = 0; i < timestamps.size(); i++) {
            JSONArray da = values.getJSONArray(i);
            //0-data2,1-data37,2-data38,3-data39,4-data42,5-data-43,6data-44
            //当前时间数据
            float[] curr = {da.getFloat(0), da.getFloat(fieldList.indexOf("Capacity_data_34")), da.getFloat(fieldList.indexOf("Formula_data_24"))};
            int[] currPan = {da.getInt(fieldList.indexOf("Capacity_data_15")), da.getInt(fieldList.indexOf("Capacity_data_16")), da.getInt(fieldList.indexOf("Capacity_data_17")),
                    da.getInt(fieldList.indexOf("Capacity_data_18")), da.getInt(fieldList.indexOf("Capacity_data_19"))};
            int curr48 = da.getInt(7);

            if (i == 0) {
                //第一次数据是上次最后一条，只做记录用，不做处理
                first = curr.clone();
                last = curr.clone();
                lastPanHead = currPan.clone();
                lastMkz = da.getInt(8);
                lastFk = da.getFloat(9);
                last48 = curr48;
                startKwh = da.getFloat(fieldList.indexOf("Capacity_data_34"));
                continue;
            }
            //计算盘头
            calcPan(currPan, lastPanHead, panHeadInfo, timestamps.getLong(i));
            //计算规格米长
            if (last[2] != curr[2]) {
                calcSpec(total[0], lastSpecLength, last[2], lastMkz, specList);
                lastSpecLength = total[0];
            }
            //处理电量跳点
            float currKwh = da.getFloat(fieldList.indexOf("Capacity_data_34"));
            if (startKwh == 0) {
                startKwh = currKwh;
            }

            if (currKwh != 0f) {
                endKwh = currKwh;
            }

//            for (int j = 0; j < first.length; j++) {
            //如果当前值为小于上一个，且上一个值不为0，则计算
            //因为会出现数据波动，停机再启动之后的第一个点不为0，为0.00几几几的
            //计算累加类的数据
            //这里只计算米长
            if (curr[0] < last[0] && last[0] != 0f) {
                calcTotal(0, last, first, total, lastMkz, lastFk);
                first[0] = curr[0];
            }
//            }
            calcAlarms(values, old, timestamps.getLong(i), alarmRecord, i);
            calcStops(curr48, last48, timestamps.getLong(i), total, stopRecord, old, i);

            //复制数组,设置last值为当前值
            last = curr.clone();
            lastPanHead = currPan.clone();
            lastMkz = da.getInt(8);
            lastFk = da.getFloat(9);
            last48 = curr48;
        }
        //最后再补一次计算
//        for (int j = 0; j < first.length; j++) {
//            //这里米长和电量都要计算
//            calcTotal(j, last, first, total, lastMkz, lastFk);
//        }
        //还是只计算米长
        calcTotal(0, last, first, total, lastMkz, lastFk);
        calcSpec(total[0], lastSpecLength, last[2], lastMkz, specList);
        //计算电量
        total[1] = endKwh - startKwh;

        //存入最后一条记录的停机状态
//        CacheUtils.put(Constants.IOT_TOKEN, table, values.getJSONArray(values.size() - 1));
        long openTime = (endTime - startTime) / 1000 + 1;
        if (openTime < total[4]) {
            total[4] = openTime;
        }
        total[3] = openTime - total[4];

        Map<String, Object> result = new HashMap<>(16);
        result.put("total", total);
        result.put("stopRecord", stopRecord);
        result.put("alarmRecord", alarmRecord);
        result.put("panHead", panHeadInfo);
        result.put("specList", specList);
        return result;
    }

    /**
     * 按规格计算米长
     *
     * @param len         当前米长
     * @param lastLen     上次记录时米长
     * @param lastDensity 密度
     * @param lastMkz     米克重
     * @param specList    列表
     */
    private void calcSpec(float len, float lastLen, float lastDensity, int lastMkz, List<TwinCalcHourSpec> specList) {
        //计算规格米长
        //如果密度有变化，就记录下来
        TwinCalcHourSpec spec = new TwinCalcHourSpec();
        spec.setDensity(BigDecimal.valueOf(lastDensity));
        float len1 = len - lastLen;
        spec.setLength(BigDecimal.valueOf(len1));
        spec.setMick(lastMkz);
        specList.add(spec);
    }

    /**
     * 计算盘头信息
     * 数组 分别是GB1-GB5
     *
     * @param currPan     当前剩余圈数
     * @param lastPanHead 上一条记录剩余圈数
     * @param panHeadInfo info
     * @param time        时间戳
     */
    private void calcPan(int[] currPan, int[] lastPanHead, List<TwinPanHeadInfo> panHeadInfo, Long time) {
        //如果当前记录大于上一条记录，则证明是重新叫料了，需要记录下当前值
        for (int i = 0; i < currPan.length; i++) {
            if (currPan[i] > lastPanHead[i]) {
                TwinPanHeadInfo info = new TwinPanHeadInfo();
                info.setRecordTime(new Date(time));
                info.setPhNum((long) (i + 1));
                info.setPhMax((long) currPan[i]);
                panHeadInfo.add(info);
            }
        }
    }


    /**
     * 停机次数计算
     * 停机处理
     *
     * @param curr       当前值
     * @param last       上一条记录的值
     * @param time       时间
     * @param total      统计
     * @param stopRecord 记录
     * @param oldV       上次的最后一条记录的值
     * @param i          当前时序
     */
    private void calcStops(int curr, int last, long time, float[] total, List<TwinRecordStop> stopRecord, JSONArray oldV, int i) {
        //取消毫秒
        time = time / 1000 * 1000;
        TwinRecordStop stop = new TwinRecordStop();
        //如果第一条记录为0，需要判断之前有木有没有结束时间的停机
        //暂时取消这个判断
//        if (i == 1 && oldV != null) {
//            int old = oldV.getInt(7);
//            if (last != old) {
//                stop.setStopType(old);
//                stop.setEndTime(new Date(time));
//                stopRecord.add(stop);
//                return;
//            }
//        }
        //如果第一条记录不为0，则直接记录停机开始时间
        if (i == 1 && curr != 0) {
            stop = new TwinRecordStop();
            //记录停机开始时间
            stop.setStopType(curr);
            stop.setStartTime(new Date(time));
            stopRecord.add(stop);
            return;
        }
        if (curr != 0) {
            total[4]++;
            if (curr != last) {
                if (last != 0) {
                    //记录上个停机类型的结束时间
                    stop.setStopType(last);
                    stop.setEndTime(new Date(time));
                    stopRecord.add(stop);
                }
                stop = new TwinRecordStop();
                //记录停机开始时间
                stop.setStopType(curr);
                stop.setStartTime(new Date(time));
                stopRecord.add(stop);
            }
        } else if (last != 0) {
            //判断停机结束之后的开机
            stop.setStopType(last);
            stop.setEndTime(new Date(time));
            stopRecord.add(stop);
        }
    }

    /**
     * 告警计算
     *
     * @param values      所有记录
     * @param old         上次存储的
     * @param time        当前时间
     * @param alarmRecord 告警记录
     * @param i           时序
     */
    private void calcAlarms(JSONArray values, JSONArray old, long time, List<TwinRecordAlarms> alarmRecord, int i) {
        JSONArray curr = values.getJSONArray(i);
        JSONArray last = values.getJSONArray(i - 1);
        TwinRecordAlarms recordAlarms;
        int index = 26;
        /*暂时取消
         //        if (i == 1 && old != null) {
         //            //判断第一条记录为0，并且上一轮存储的值，需要判断之前有没有未结束的告警
         //            int j = 0;
         //            for (; j < index; j++) {
         //                boolean oldV = curr.getBool(j + 12);
         //                boolean lastV = last.getBool(j + 12);
         //                if (oldV && !lastV) {
         //                    recordAlarms = new TwinRecordAlarms();
         //                    recordAlarms.setAlarmType(j + 1);
         //                    recordAlarms.setEndTime(new Date(time));
         //                    alarmRecord.add(recordAlarms);
         //                }
         //            }
         //            //需要单独计算alarm27
         //            //数据位置
         //            int pos = j + 12;
         //            int old27 = curr.getInt(pos);
         //            //上面已经处理过i=0,所以这里i不可能等于0
         //            int last27 = last.getInt(pos);
         //            if (old27 != 0 && last27 == 0) {
         //                recordAlarms = new TwinRecordAlarms();
         //                recordAlarms.setAlarmType(j + 1);
         //                recordAlarms.setEndTime(new Date(time));
         //                alarmRecord.add(recordAlarms);
         //            }
         //        }
         */
        boolean[] flags = new boolean[index + 1];
        Arrays.fill(flags, true);
        if (i == 1) {
            int j = 0;
            for (; j < index; j++) {
                //允许的告警编号
                if (Tools.findAllow(j + 1)) {
                    boolean currV = curr.getBool(j + 12);
                    if (currV) {
                        flags[j] = false;
                        recordAlarms = new TwinRecordAlarms();
                        recordAlarms.setAlarmType(j + 1);
                        recordAlarms.setStartTime(new Date(time));
                        alarmRecord.add(recordAlarms);
                    }
                }
            }
            //需要单独计算alarm27
            //数据位置
            int pos = j + 12;
            int curr27 = curr.getInt(pos);
            if (curr27 != 0) {
                flags[j] = false;
                recordAlarms = new TwinRecordAlarms();
                recordAlarms.setAlarmType(j + 1);
                recordAlarms.setStartTime(new Date(time));
                alarmRecord.add(recordAlarms);
            }
        }
        int j = 0;
        for (; j < index; j++) {
            //允许的告警编号
            if (Tools.findAllow(j + 1)) {
                if (flags[j]) {
                    boolean currV = curr.getBool(j + 12);
                    boolean lastV = last.getBool(j + 12);
                    if (currV && !lastV) {
                        recordAlarms = new TwinRecordAlarms();
                        recordAlarms.setAlarmType(j + 1);
                        recordAlarms.setStartTime(new Date(time));
                        alarmRecord.add(recordAlarms);
                    }
                    if (!currV && lastV) {
                        recordAlarms = new TwinRecordAlarms();
                        recordAlarms.setAlarmType(j + 1);
                        recordAlarms.setEndTime(new Date(time));
                        alarmRecord.add(recordAlarms);
                    }
                }
            }
        }
        if (flags[j]) {
            //需要单独计算alarm27
            //数据位置
            int pos = j + 12;
            int alarm27 = curr.getInt(pos);
            //上面已经处理过i=0,所以这里i不可能等于0
            int last27 = last.getInt(pos);
            if (alarm27 != 0 && last27 == 0) {
                recordAlarms = new TwinRecordAlarms();
                recordAlarms.setAlarmType(j + 1);
                recordAlarms.setStartTime(new Date(time));
                alarmRecord.add(recordAlarms);
            }
            if (alarm27 == 0 && last27 != 0) {
                recordAlarms = new TwinRecordAlarms();
                recordAlarms.setAlarmType(j + 1);
                recordAlarms.setEndTime(new Date(time));
                alarmRecord.add(recordAlarms);
            }
        }
    }

    /**
     * 重量计算 提取公共方法
     */
    private void calcTotal(int j, float[] last, float[] first, float[] total, int lastMkz, float lastFk) {
        float v = last[j] - first[j];
        total[j] += v;
        if (j == 0) {
            //如果是米长，则计算重量，米长*米克重*2 2024-11-15
            float weight = BigDecimal.valueOf(v * lastMkz * 2 / 1000 / 1000).setScale(2, RoundingMode.HALF_UP).floatValue();
            total[2] = total[2] + weight;
        }
    }

}