| | |
| | | import com.moral.constant.SeparateTableType; |
| | | import com.moral.util.AmendUtils; |
| | | import com.moral.util.DateUtils; |
| | | |
| | | import com.moral.util.MybatisPLUSUtils; |
| | | |
| | | import lombok.extern.slf4j.Slf4j; |
| | | import org.apache.commons.httpclient.HttpClient; |
| | | import org.apache.commons.httpclient.methods.PostMethod; |
| | | 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.util.ObjectUtils; |
| | | |
| | | import java.io.*; |
| | | import java.math.BigDecimal; |
| | | import java.net.InetAddress; |
| | | import java.net.InetSocketAddress; |
| | | import java.net.Socket; |
| | | import java.net.SocketAddress; |
| | | import java.nio.charset.StandardCharsets; |
| | | import java.util.*; |
| | | import java.util.concurrent.atomic.AtomicInteger; |
| | | import java.util.function.Supplier; |
| | |
| | | |
| | | } |
| | | |
| | | /** |
| | | * 计算疾控中心的小时数据 |
| | | */ |
| | | @Override |
| | | public void dateInsertHistoryHourlyJKAvg(String start,String end) { |
| | | QueryWrapper<Device> wrapper = new QueryWrapper<>(); |
| | | wrapper.select("mac"); |
| | | wrapper.eq("organization_id",72); |
| | | wrapper.eq("is_delete",Constants.NOT_DELETE); |
| | | List<Device> devices = deviceMapper.selectList(wrapper); |
| | | ArrayList<String> macs = new ArrayList<>(); |
| | | for (Device device : devices) { |
| | | macs.add(device.getMac()); |
| | | } |
| | | //从数据库获取数据参数 |
| | | Map<String, Object> params = new HashMap<>(); |
| | | Date startTime=null; |
| | | Date endTime=null; |
| | | if (!ObjectUtils.isEmpty(start) || !ObjectUtils.isEmpty(end)){ |
| | | startTime = DateUtils.getDate(start, DateUtils.yyyy_MM_dd_HH_mm_ss_EN); |
| | | endTime = DateUtils.getDate(end, DateUtils.yyyy_MM_dd_HH_mm_ss_EN); |
| | | }else { |
| | | //时间格式化:yyyy-MM-dd HH:mm |
| | | String format = DateUtils.yyyy_MM_dd_HH_EN; |
| | | Date now = new Date(); |
| | | //开始时间 |
| | | startTime = DateUtils.dataToTimeStampTime(DateUtils.addHours(now, -1), format); |
| | | //结束时间 |
| | | endTime = DateUtils.dataToTimeStampTime(now, format); |
| | | } |
| | | |
| | | params.put("start", startTime); |
| | | params.put("end", endTime); |
| | | params.put("macs",macs); |
| | | //获取数据的分钟表后缀 |
| | | String timeUnits = DateUtils.dateToDateString(startTime, 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); |
| | | // 获取疾控中心设备的一小时内分钟的数据 |
| | | List<Map<String, Object>> hourlyData = historyMinutelyMapper.getHistoryMinutelyData(params); |
| | | if (ObjectUtils.isEmpty(hourlyData)) { |
| | | return; |
| | | } |
| | | //按mac分组 |
| | | Map<String, List<Map<String, Object>>> data = hourlyData.parallelStream() |
| | | .collect(Collectors.groupingBy(o -> (String) o.get("mac"))); |
| | | //存入数据库的结果集 |
| | | List<Map<String, Object>> insertData = new ArrayList<>(); |
| | | data.forEach((key, value) -> { |
| | | Map<String, Object> historyHourly = new HashMap<>(); |
| | | historyHourly.put("mac", key); |
| | | historyHourly.put("time",start); |
| | | |
| | | Map<String, Object> jsonMap = new HashMap<>(); |
| | | Map<String, Object> map = new HashMap<>(); |
| | | map.put("data", value); |
| | | map.put("type", "hour"); |
| | | |
| | | 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(); |
| | | AtomicInteger size = new AtomicInteger(); |
| | | 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; |
| | | } |
| | | } |
| | | size.getAndIncrement(); |
| | | return DoubleStream.of(aDouble); |
| | | }); |
| | | OptionalDouble average = optionalDouble.average(); |
| | | if (average.isPresent()) { |
| | | //银行家算法修约 |
| | | double sciCal = AmendUtils.sciCal(average.getAsDouble(), 4); |
| | | jsonMap.put(sensorCode, sciCal); |
| | | //标志位 |
| | | if (size.get() >= 45) { |
| | | jsonMap.put(sensorCode + "-" + Constants.MARKER_BIT_KEY, Constants.MARKER_BIT_TRUE); |
| | | } else { |
| | | jsonMap.put(sensorCode + "-" + Constants.MARKER_BIT_KEY, Constants.MARKER_BIT_FALSE); |
| | | } |
| | | } |
| | | }); |
| | | historyHourly.put("version", value.get(0).get("version")); |
| | | historyHourly.put("value", JSONObject.toJSONString(jsonMap)); |
| | | insertData.add(historyHourly); |
| | | }); |
| | | //存入数据库 |
| | | historyHourlyMapper.insertHistoryHourlyAvg(insertData,timeUnits); |
| | | |
| | | } |
| | | |
| | | private String strList(String startTime,String mn,String msg){ |
| | | String qn = "QN="+ startTime + "001;ST=22;CN=2061;PW=123456;MN="+mn.toUpperCase()+";CP=&&DataTime="+startTime+";"; |
| | | Map<String, Object> data = JSON.parseObject(msg, Map.class); |