package com.moral.api.service.impl; import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper; import com.googlecode.aviator.AviatorEvaluator; import com.googlecode.aviator.Expression; import com.moral.api.entity.Device; import com.moral.api.service.DeviceService; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import org.springframework.util.CollectionUtils; import org.springframework.util.ObjectUtils; import java.util.*; import java.util.stream.Collectors; import java.util.stream.DoubleStream; import com.alibaba.fastjson.JSONObject; import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; import com.moral.api.entity.Sensor; import com.moral.api.mapper.HistoryFiveMinutelyMapper; import com.moral.api.mapper.HistoryMinutelyMapper; import com.moral.api.service.HistoryFiveMinutelyService; import com.moral.api.service.SensorService; import com.moral.constant.Constants; import com.moral.constant.RedisConstants; import com.moral.util.AmendUtils; import com.moral.util.DateUtils; import org.springframework.util.StringUtils; @Service public class HistoryFiveMinutelyServiceImpl implements HistoryFiveMinutelyService { @Value("${result.device.list}") private String deviceList; @Autowired private HistoryFiveMinutelyMapper historyFiveMinutelyMapper; @Autowired private HistoryMinutelyMapper historyMinutelyMapper; @Autowired private SensorService sensorService; @Autowired private RedisTemplate redisTemplate; @Autowired private DeviceService deviceService; @Override public void createTable(String timeUnits) { historyFiveMinutelyMapper.createTable(timeUnits); } @Override @Transactional public void insertHistoryFiveMinutely() { //时间格式化:yyyy-MM-dd HH:mm String format = DateUtils.yyyy_MM_dd_HH_mm_EN; Date now = new Date(); now = DateUtils.getFiveMinuteDate(now); //从数据库获取数据参数 Map params = new HashMap<>(); //开始时间 Date start = DateUtils.dataToTimeStampTime(DateUtils.getDateOfMin(now, -5), format); //结束时间 Date end = DateUtils.dataToTimeStampTime(now, format); params.put("start", start); params.put("end", end); //获取数据的分钟表后缀 String timeUnits = DateUtils.dateToDateString(start, DateUtils.yyyyMM_EN); params.put("timeUnits", timeUnits); //因子 QueryWrapper sensorQueryWrapper = new QueryWrapper<>(); sensorQueryWrapper.select("code", "lower", "upper").eq("is_delete", Constants.NOT_DELETE); List sensors = sensorService.list(sensorQueryWrapper); //获取所有设备的5分钟内的数据 List> fiveMinutelyData = historyMinutelyMapper.getHistoryMinutelyData(params); if (fiveMinutelyData.size() == 0) { return; } listDistDeviceState(params); //按mac分组 Map>> data = fiveMinutelyData.parallelStream() .collect(Collectors.groupingBy(o -> (String) o.get("mac"))); //存入数据库的结果集 List> insertData = new ArrayList<>(); data.forEach((key, value) -> { Map dataMap = new HashMap<>(); dataMap.put("mac", key); dataMap.put("time", start); Map jsonMap = new HashMap<>(); Map map = new HashMap<>(); map.put("data", value); map.put("type", "fiveMinutes"); for (Sensor sensor : sensors) { String sensorCode = sensor.getCode(); //风向上下限 if (sensorCode.equals(Constants.SENSOR_CODE_WIND_DIR)) { if (sensor.getUpper() != null) { map.put("windDirUpper", sensor.getUpper()); } if (sensor.getLower() != null) { map.put("windDirLower", sensor.getLower()); } } //风速上下限 if (sensorCode.equals(Constants.SENSOR_CODE_WIND_SPEED)) { if (sensor.getUpper() != null) { map.put("windSpeedUpper", sensor.getUpper()); } if (sensor.getLower() != null) { map.put("windSpeedLower", sensor.getLower()); } } } //风向均值计算并修约 Map windDirAvg = AmendUtils.getWindDirAvg(map); if (!ObjectUtils.isEmpty(windDirAvg)) { jsonMap.putAll(windDirAvg); } //除风向外其他因子均值计算 sensors.forEach(sensor -> { String sensorCode = sensor.getCode(); Double upper = sensor.getUpper(); Double lower = sensor.getLower(); DoubleStream optionalDouble = value.parallelStream() .flatMapToDouble(v -> { Map dataValue = JSONObject.parseObject((String) v.get("value"), Map.class); Object sensorValue = dataValue.get(sensorCode); //数据有效性标记位 Object flag = dataValue.get(sensorCode + "-" + Constants.MARKER_BIT_KEY); if (!Constants.MARKER_BIT_TRUE.equals(flag)) { return null; } if (ObjectUtils.isEmpty(sensorValue)) { return null; } //风向单独计算 if (Constants.SENSOR_CODE_WIND_DIR.equals(sensorCode)) { return null; } //剔除数据超过上下限的数据 double aDouble = Double.parseDouble(sensorValue.toString()); if (!ObjectUtils.isEmpty(upper)) { if (aDouble > upper) { return null; } } if (!ObjectUtils.isEmpty(lower)) { if (aDouble < lower) { return null; } } return DoubleStream.of(aDouble); }); OptionalDouble average = optionalDouble.average(); if (average.isPresent()) { //银行家算法修约 double sciCal = AmendUtils.sciCal(average.getAsDouble(), 4); jsonMap.put(sensorCode, sciCal); } }); dataMap.put("value", JSONObject.toJSONString(jsonMap)); //存入redis jsonMap.put("dataTime", DateUtils.dateToDateString(start)); redisTemplate.opsForHash().put(RedisConstants.DATA_FIVE_MINUTES, key, jsonMap); insertData.add(dataMap); judgeDeviceState(key,jsonMap); }); //5分钟表后缀 String insertTimeUnits = DateUtils.dateToDateString(start, DateUtils.yyyyMM_EN); //存入数据库 historyFiveMinutelyMapper.insertHistoryFiveMinutely(insertData, insertTimeUnits); } // 修改设备状态 private void judgeDeviceState(String mac,Map dataMap){ List sensorList = Arrays.asList("a34004","a34002","a21026","a21004","a21005","a05024","a99054"); Device device = (Device) redisTemplate.opsForHash().get(RedisConstants.DEVICE_INFO, mac); if(Objects.isNull(device)){ return; } List sensors = device.getVersion().getSensors(); Expression expression; int state = 1; for (Sensor sensor : sensors) { if(!sensorList.contains(sensor.getCode())){ continue; } //因子报警等级 String alarmLevel = sensor.getAlarmLevel(); if (StringUtils.isEmpty(alarmLevel)) { continue; } List list = JSONObject.parseObject(alarmLevel, List.class); String sensorCode = sensor.getCode(); //转换公式 String formula = sensor.getFormula(); //设备所传因子与配置型号因子不一致的处理逻辑 if (dataMap.get(sensorCode) != null) { String sensorValue = String.valueOf(dataMap.get(sensorCode)); double value = Double.parseDouble(sensorValue); if (formula != null) { //转换后因子值 sensorValue = formula.replace("{0}", sensorValue); expression = AviatorEvaluator.compile(sensorValue); value = Double.parseDouble(expression.execute().toString()); } int sensorState = judgeState(list, value); if (sensorState > state) { state = sensorState; } } } //修改设备状态 Device devices = new Device(); devices.setId(device.getId()); devices.setState(String.valueOf(state)); deviceService.updateById(devices); } private int judgeState(List levels, Double data) { int state = 1; for (int i = levels.size() - 1; i >= 0; i--) { Double level = Double.parseDouble(levels.get(i).toString()); if (data >= level) { state = i + 2; break; } } return state; } private void listDistDeviceState(Map params){ QueryWrapper queryWrapper = new QueryWrapper<>(); queryWrapper.eq("is_delete",Constants.NOT_DELETE); //获取所有在线设备 List devices = deviceService.list(queryWrapper); List macOldList = devices.stream().map(Device::getMac).collect(Collectors.toList()); List macList = historyMinutelyMapper.getHistoryMinutelyMacData(params); if(!CollectionUtils.isEmpty(macOldList)){ List listDictDevice = Arrays.asList(deviceList.split(",")); List list2 = macOldList.stream().filter(e -> !listDictDevice.contains(e)).collect(Collectors.toList()); List list3 = list2.stream().filter(e -> !macList.contains(e)).collect(Collectors.toList()); List idList = devices.stream().filter(e->list3.contains(e.getMac())).map(Device::getId).collect(Collectors.toList()); for(Integer i : idList){ Device device = new Device(); device.setId(i); device.setState(Constants.DEVICE_STATE_OFFLINE); deviceService.updateById(device); } } } @Override public void insertHistoryFiveMinutely(String yz, String mac) { String format = DateUtils.yyyy_MM_dd_HH_mm_EN; Date now = new Date(); now = DateUtils.getFiveMinuteDate(now); //从数据库获取数据参数 Map params = new HashMap<>(); //开始时间 Date start = DateUtils.dataToTimeStampTime(DateUtils.getDateOfMin(now, -5), format); //结束时间 Date end = DateUtils.dataToTimeStampTime(now, format); params.put("start", start); params.put("end", end); //获取数据的分钟表后缀 String timeUnits = DateUtils.dateToDateString(start, DateUtils.yyyyMM_EN); params.put("timeUnits", timeUnits); /*if(!StringUtils.isEmpty(mac)){ params.put("macs", Arrays.asList(mac)); }*/ //因子 QueryWrapper sensorQueryWrapper = new QueryWrapper<>(); sensorQueryWrapper.select("code", "lower", "upper").eq("is_delete", Constants.NOT_DELETE); /*if(!StringUtils.isEmpty(yz)){ sensorQueryWrapper.eq("code",yz); }*/ List sensors = sensorService.list(sensorQueryWrapper); //获取所有设备的5分钟内的数据 List> fiveMinutelyData = historyMinutelyMapper.getHistoryMinutelyData(params); if (fiveMinutelyData.size() == 0) { return; } listDistDeviceState(params); //按mac分组 Map>> data = fiveMinutelyData.parallelStream() .collect(Collectors.groupingBy(o -> (String) o.get("mac"))); //存入数据库的结果集 List> insertData = new ArrayList<>(); data.forEach((key, value) -> { Map dataMap = new HashMap<>(); dataMap.put("mac", key); dataMap.put("time", start); Map jsonMap = new HashMap<>(); Map map = new HashMap<>(); map.put("data", value); map.put("type", "fiveMinutes"); for (Sensor sensor : sensors) { String sensorCode = sensor.getCode(); //风向上下限 if (sensorCode.equals(Constants.SENSOR_CODE_WIND_DIR)) { if (sensor.getUpper() != null) { map.put("windDirUpper", sensor.getUpper()); } if (sensor.getLower() != null) { map.put("windDirLower", sensor.getLower()); } } //风速上下限 if (sensorCode.equals(Constants.SENSOR_CODE_WIND_SPEED)) { if (sensor.getUpper() != null) { map.put("windSpeedUpper", sensor.getUpper()); } if (sensor.getLower() != null) { map.put("windSpeedLower", sensor.getLower()); } } } //风向均值计算并修约 Map windDirAvg = AmendUtils.getWindDirAvg(map); if (!ObjectUtils.isEmpty(windDirAvg)) { jsonMap.putAll(windDirAvg); } //除风向外其他因子均值计算 sensors.forEach(sensor -> { String sensorCode = sensor.getCode(); Double upper = sensor.getUpper(); Double lower = sensor.getLower(); DoubleStream optionalDouble = value.parallelStream() .flatMapToDouble(v -> { Map dataValue = JSONObject.parseObject((String) v.get("value"), Map.class); Object sensorValue = dataValue.get(sensorCode); //数据有效性标记位 Object flag = dataValue.get(sensorCode + "-" + Constants.MARKER_BIT_KEY); if (!Constants.MARKER_BIT_TRUE.equals(flag)) { return null; } if (ObjectUtils.isEmpty(sensorValue)) { return null; } //风向单独计算 if (Constants.SENSOR_CODE_WIND_DIR.equals(sensorCode)) { return null; } //剔除数据超过上下限的数据 double aDouble = Double.parseDouble(sensorValue.toString()); if (!ObjectUtils.isEmpty(upper)) { if (aDouble > upper) { return null; } } if (!ObjectUtils.isEmpty(lower)) { if (aDouble < lower) { return null; } } return DoubleStream.of(aDouble); }); OptionalDouble average = optionalDouble.average(); if (average.isPresent()) { //银行家算法修约 double sciCal = AmendUtils.sciCal(average.getAsDouble(), 4); jsonMap.put(sensorCode, sciCal); } }); dataMap.put("value", JSONObject.toJSONString(jsonMap)); //存入redis jsonMap.put("dataTime", DateUtils.dateToDateString(start)); redisTemplate.opsForHash().put(RedisConstants.DATA_FIVE_MINUTES, key, jsonMap); insertData.add(dataMap); System.out.println(">>>"+key+"->>>"+jsonMap.toString()); judgeDeviceState(key,jsonMap); }); //5分钟表后缀 // String insertTimeUnits = DateUtils.dateToDateString(start, DateUtils.yyyyMM_EN); //存入数据库 // historyFiveMinutelyMapper.insertHistoryFiveMinutely(insertData, insertTimeUnits); } }