潘钊 / bsth_control · Files

package com.bsth.data.gpsdata;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bsth.Application;
import com.bsth.data.BasicData;
import com.bsth.data.forecast.ForecastRealServer;
import com.bsth.data.schedule.DayOfSchedule;
import com.bsth.entity.realcontrol.ScheduleRealInfo;
import com.bsth.util.ConfigUtil;
import com.google.common.collect.TreeMultimap;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpEntity;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 *
 * @ClassName: GpsRealData
 * @Description: TODO(实时GPS数据集合)
 * @author PanZhao
 * @date 2016年8月12日 下午2:04:41
 *
 */
@Component
public class GpsRealData implements CommandLineRunner{

	static Logger logger = LoggerFactory.getLogger(GpsRealData.class);

	private static Map<String, GpsEntity> gpsMap;

	//按线路分组设备号
	private static TreeMultimap<String, String> lineCode2Devices;

	// 网关数据接口地址
	private static String url;

	@Autowired
	GpsDataLoader gpsDataLoader;

	@Autowired
	DayOfSchedule dayOfSchedule;

	@Autowired
	ForecastRealServer forecastRealServer;
	/**
	 * 构造函数
	 */
	public GpsRealData(){
		gpsMap = new HashMap<>();
		lineCode2Devices = TreeMultimap.create();
		url = ConfigUtil.get("http.gps.real.url");
	}

	@Override
	public void run(String... arg0) throws Exception {
		logger.info("gpsDataLoader,20,6");
		Application.mainServices.scheduleWithFixedDelay(gpsDataLoader, 20, 5, TimeUnit.SECONDS);
	}

	public GpsEntity add(GpsEntity gps) {
		String device = gps.getDeviceId();
		GpsEntity old = gpsMap.get(device);

		if(!StringUtils.isEmpty(gps.getStopNo())){
			//定时定距数据附带站点编码改变
			if(null == old || !gps.getStopNo().equals(old.getStopNo())){
				gps.setArrTime(gps.getTimestamp());
				//预测到达终点时间
				forecastRealServer.forecast(gps.getNbbm(), gps);
			}
			else{
				gps.setArrTime(old.getArrTime());
				//不预测， 重新计算终点时间
				gps.setExpectStopTime(forecastRealServer.expectStopTime(gps.getNbbm()));
			}
		}

		gpsMap.put(device, gps);
		if(StringUtils.isNotBlank(gps.getLineId()))
			lineCode2Devices.put(gps.getLineId(), device);
		return gps;
	}

	/**
	 *
	 * @Title: get @Description: TODO(设备号获取GPS)
	 */
	public GpsEntity get(String deviceId) {
		return gpsMap.get(deviceId);
	}

	/**
	 *
	 * @Title: get @Description: TODO(线路编码获取GPS集合) @throws
	 */
	public List<GpsEntity> getByLine(String lineCode) {
		NavigableSet<String> set = lineCode2Devices.get(lineCode);

		List<GpsEntity> rs = new ArrayList<>();
		GpsEntity gps;
		ScheduleRealInfo sch;
		for(String device : set){
			gps = gpsMap.get(device);
			//过滤异常GPS数据
			if(gps.isAbnormal())
				continue;

			sch = dayOfSchedule.execPlamMap().get(gps.getNbbm());
			if(null != sch)
				gps.setSchId(sch.getId());
			rs.add(gps);
		}

		return rs;
	}

	public List<GpsEntity> get(List<String> pArray){
		List<GpsEntity> list = new ArrayList<>();

		for(String code : pArray)
			list.addAll(getByLine(code));
		return list;
	}

	public Set<String> allDevices(){
		return gpsMap.keySet();
	}

	public GpsEntity findByDeviceId(String deviceId) {
		return gpsMap.get(deviceId);
	}

	public Collection<GpsEntity> all(){
		return gpsMap.values();
	}

	public void remove(String device){
		gpsMap.remove(device);
	}
	@Component
	public static class GpsDataLoader extends Thread{

		Logger logger = LoggerFactory.getLogger(GpsDataLoader.class);

		@Autowired
		GpsRealData gpsRealData;

		@Override
		public void run() {
			try{
				load();
			}catch(Exception e){
				logger.error("", e);
			}
		}

		public void load() throws Exception {
			List<GpsEntity> list = new ArrayList<>();
			CloseableHttpClient httpClient = null;
			CloseableHttpResponse response = null;
			try {
				httpClient = HttpClients.createDefault();
				HttpGet get = new HttpGet(url);

				response = httpClient.execute(get);

				HttpEntity entity = response.getEntity();
				if (null != entity) {
					// 返回数据量比较大，建议以流的形式读取
					BufferedReader br = new BufferedReader(new InputStreamReader(entity.getContent()));
					StringBuffer stringBuffer = new StringBuffer();
					String str = "";
					while ((str = br.readLine()) != null)
						stringBuffer.append(str);

					JSONObject jsonObj = JSON.parseObject(stringBuffer.toString());

					if (jsonObj != null)
						list = JSON.parseArray(jsonObj.getString("data"), GpsEntity.class);

					String nbbm;
					for(GpsEntity gps : list){
						//没有设备号
						if(StringUtils.isBlank(gps.getDeviceId()))
							continue;

						nbbm = BasicData.deviceId2NbbmMap.get(gps.getDeviceId());
						if(StringUtils.isBlank(nbbm))
							gps.setAbnormal(true);//标记为异常数据
						else
							gps.setNbbm(nbbm);

						gps.setStationName(BasicData.stationCode2NameMap.get(gps.getStopNo()));
						gpsRealData.add(gps);

						//纠正走向
						correctUpdown(gps);
						/*if(issEPoint(gps))
							continue;

						//如果走向未知，尝试根据站点纠正走向
						if(gps.getUpDown() == -1){
							updown=stationUpDownMap.get(gps.getLineId()+"_"+gps.getStopNo());
							if(updown != null)
								gps.setUpDown(updown);
						}
						//如果站点编码和走向相反（即上行站点ID，走向为下行），尝试根据站点纠正走向
						updown=stationUpDownMap.get(gps.getLineId()+"_"+gps.getStopNo());
						if(updown != null && !gps.getUpDown().equals(updown)){
							gps.setUpDown(updown);
						}*/
					}
				} else
					logger.error("result is null");
			} catch(Exception e){
				logger.error("", e);
			}
			finally {
				if (null != httpClient)
					httpClient.close();
				if(null != response)
					response.close();
			}
		}

		/**
		 * 是否是起终点
		 * @param gps
		 * @return
		 */
		public boolean isSEPoint(GpsEntity gps){
			String key = gps.getLineId()+"_"+gps.getUpDown()+"_"
					,stationCode;

			if(BasicData.lineSEPointMap.containsKey(key+"B")){
				stationCode = BasicData.lineSEPointMap.get(key+"B");
				if(gps.getStopNo().equals(stationCode)){
					gps.setsEPoint(true);
					return true;
				}
			}

			if(BasicData.lineSEPointMap.containsKey(key+"E")){
				stationCode = BasicData.lineSEPointMap.get(key+"E");
				if(gps.getStopNo().equals(stationCode)){
					gps.setsEPoint(true);
					return true;
				}
			}
			return false;
		}

		/**
		 * 纠正上下行
		 * @param gps
		 */
		public void correctUpdown(GpsEntity gps){
			Integer updown=BasicData.lineStationUpDownMap.get(gps.getLineId()+"_"+gps.getStopNo());
			if(updown != null && !updown.equals(gps.getUpDown()))
				gps.setUpDown(updown);
			/*//如果走向未知，尝试根据站点纠正走向
			if(gps.getUpDown() == -1){

			}*/

			if(isSEPoint(gps))
				return;

			/*//如果站点编码和走向相反（即上行站点ID，走向为下行），尝试根据站点纠正走向
			updown=BasicData.lineStationUpDownMap.get(gps.getLineId()+"_"+gps.getStopNo());
			if(updown != null && !gps.getUpDown().equals(updown)){
				gps.setUpDown(updown);
			}*/
		}
	}
}