|  |  |  | 
|---|
|  |  |  | package com.moral.api.service.impl; | 
|---|
|  |  |  |  | 
|---|
|  |  |  | import org.springframework.beans.factory.annotation.Autowired; | 
|---|
|  |  |  | import org.springframework.data.redis.core.RedisTemplate; | 
|---|
|  |  |  | import org.springframework.stereotype.Service; | 
|---|
|  |  |  | import org.springframework.transaction.annotation.Transactional; | 
|---|
|  |  |  | import org.springframework.util.ObjectUtils; | 
|---|
|  |  |  |  | 
|---|
|  |  |  | import java.text.SimpleDateFormat; | 
|---|
|  |  |  | import java.util.ArrayList; | 
|---|
|  |  |  | import java.util.Date; | 
|---|
|  |  |  | import java.util.HashMap; | 
|---|
|  |  |  | import java.util.List; | 
|---|
|  |  |  | import java.util.Map; | 
|---|
|  |  |  | import java.util.OptionalDouble; | 
|---|
|  |  |  | import java.util.stream.Collectors; | 
|---|
|  |  |  | import java.util.stream.DoubleStream; | 
|---|
|  |  |  |  | 
|---|
|  |  |  | import com.alibaba.fastjson.JSON; | 
|---|
|  |  |  | 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.HistoryMinutelyService; | 
|---|
|  |  |  | 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; | 
|---|
|  |  |  |  | 
|---|
|  |  |  | @Service | 
|---|
|  |  |  | 
|---|
|  |  |  | private HistoryFiveMinutelyMapper historyFiveMinutelyMapper; | 
|---|
|  |  |  |  | 
|---|
|  |  |  | @Autowired | 
|---|
|  |  |  | private HistoryMinutelyService historyMinutelyService; | 
|---|
|  |  |  | private HistoryMinutelyMapper historyMinutelyMapper; | 
|---|
|  |  |  |  | 
|---|
|  |  |  | @Autowired | 
|---|
|  |  |  | private SensorService sensorService; | 
|---|
|  |  |  |  | 
|---|
|  |  |  | @Autowired | 
|---|
|  |  |  | private RedisTemplate redisTemplate; | 
|---|
|  |  |  |  | 
|---|
|  |  |  | @Override | 
|---|
|  |  |  | public void createTable(String 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<String, Object> params = new HashMap<>(); | 
|---|
|  |  |  | //开始时间(分钟) | 
|---|
|  |  |  | Date start = DateUtils.convertDate(DateUtils.getDateOfMin(-5)); | 
|---|
|  |  |  | //结束时间(分钟) | 
|---|
|  |  |  | Date end = DateUtils.convertDate(new Date()); | 
|---|
|  |  |  | //开始时间 | 
|---|
|  |  |  | 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<Sensor> sensorQueryWrapper = new QueryWrapper<>(); | 
|---|
|  |  |  | sensorQueryWrapper.select("code", "lower", "upper").eq("is_delete", Constants.NOT_DELETE); | 
|---|
|  |  |  | List<Sensor> sensors = sensorService.list(sensorQueryWrapper); | 
|---|
|  |  |  |  | 
|---|
|  |  |  | params.put("sensorKeys", null); | 
|---|
|  |  |  | //获取所有设备的5分钟内的数据 | 
|---|
|  |  |  | List<Map<String, Object>> fiveMinutelyData = historyMinutelyMapper.getHistoryMinutelyData(params); | 
|---|
|  |  |  | if (fiveMinutelyData.size() == 0) { | 
|---|
|  |  |  | return; | 
|---|
|  |  |  | } | 
|---|
|  |  |  |  | 
|---|
|  |  |  | historyMinutelyService.getHistoryFiveMinutelyData(params); | 
|---|
|  |  |  | /*List<Map<String, Object>> list = new ArrayList<>(); | 
|---|
|  |  |  | Map<String, Object> map = new HashMap<>(); | 
|---|
|  |  |  | map.put("mac", "123456"); | 
|---|
|  |  |  | Date date = new Date(); | 
|---|
|  |  |  | map.put("time", date); | 
|---|
|  |  |  | Map<String, Object> value = new HashMap<>(); | 
|---|
|  |  |  | value.put("e1", 10); | 
|---|
|  |  |  | value.put("e2", 20); | 
|---|
|  |  |  | map.put("value", JSON.toJSONString(value)); | 
|---|
|  |  |  | list.add(map);*/ | 
|---|
|  |  |  | //按mac分组 | 
|---|
|  |  |  | Map<String, List<Map<String, Object>>> data = fiveMinutelyData.parallelStream() | 
|---|
|  |  |  | .collect(Collectors.groupingBy(o -> (String) o.get("mac"))); | 
|---|
|  |  |  |  | 
|---|
|  |  |  | historyFiveMinutelyMapper.insertHistoryFiveMinutely(null, null); | 
|---|
|  |  |  | //存入数据库的结果集 | 
|---|
|  |  |  | List<Map<String, Object>> insertData = new ArrayList<>(); | 
|---|
|  |  |  |  | 
|---|
|  |  |  | data.forEach((key, value) -> { | 
|---|
|  |  |  | Map<String, Object> dataMap = new HashMap<>(); | 
|---|
|  |  |  | dataMap.put("mac", key); | 
|---|
|  |  |  | dataMap.put("time", start); | 
|---|
|  |  |  | Map<String, Object> jsonMap = new HashMap<>(); | 
|---|
|  |  |  |  | 
|---|
|  |  |  |  | 
|---|
|  |  |  | Map<String, Object> 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<String, Object> 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<String, Object> 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 | 
|---|
|  |  |  | redisTemplate.opsForHash().put(RedisConstants.DATA_FIVE_MINUTES, key, jsonMap); | 
|---|
|  |  |  | insertData.add(dataMap); | 
|---|
|  |  |  | }); | 
|---|
|  |  |  |  | 
|---|
|  |  |  | //5分钟表后缀 | 
|---|
|  |  |  | String insertTimeUnits = DateUtils.dateToDateString(now, DateUtils.yyyyMM_EN); | 
|---|
|  |  |  | //存入数据库 | 
|---|
|  |  |  | historyFiveMinutelyMapper.insertHistoryFiveMinutely(insertData, insertTimeUnits); | 
|---|
|  |  |  | } | 
|---|
|  |  |  | } | 
|---|