Selaa lähdekoodia

网关号启用后带来的pointset命令bug修复;
楼号各种转换情况带来的下发及统计bug修复。

Jay Lee 6 vuotta sitten
vanhempi
commit
f48bf62704

+ 26 - 12
collect/src/main/java/com/saga/test/ShenQI.java

@@ -3,7 +3,6 @@ package com.saga.test;
 import java.io.FileInputStream;
 import java.io.FileOutputStream;
 import java.io.OutputStream;
-import java.text.ParseException;
 import java.util.Calendar;
 import java.util.Date;
 import java.util.HashMap;
@@ -48,18 +47,33 @@ public class ShenQI {
 
 			GetPointList(agent, "db_time_data", "dy_pointlist", building);
 
+			// String timefrom = "20171001000000";
+			// String timeto = "20171101000000";
+			// export2(timefrom, timeto);
+			// timefrom = "20171101000000";
+			// timeto = "20171201000000";
+			// export2(timefrom, timeto);
+			// timefrom = "20180301000000";
+			// timeto = "20180401000000";
+			// export2(timefrom, timeto);
+			// timefrom = "20180701000000";
+			// timeto = "20180801000000";
+			// export2(timefrom, timeto);
 			String timefrom = "20171001000000";
-			String timeto = "20171101000000";
-			export2(timefrom, timeto);
-			timefrom = "20171101000000";
-			timeto = "20171201000000";
-			export2(timefrom, timeto);
-			timefrom = "20180301000000";
-			timeto = "20180401000000";
-			export2(timefrom, timeto);
-			timefrom = "20180701000000";
-			timeto = "20180801000000";
-			export2(timefrom, timeto);
+			String timeto = "20180814000000";
+			Calendar calendarFrom = Calendar.getInstance();
+			Calendar calendarTo = Calendar.getInstance();
+			Date from = Constant.parse(timefrom);
+			Date to = Constant.parse(timeto);
+			calendarFrom.setTime(from);
+			calendarTo.setTime(to);
+			while (calendarFrom.before(calendarTo)) {
+				Date datetmp = calendarFrom.getTime();
+				calendarFrom.add(Calendar.MONTH, 1);
+
+				export2(Constant.format(datetmp), Constant.format(calendarFrom.getTime()));
+			}
+			// export2(timefrom, timeto);
 
 		} catch (Exception e) {
 			// TODO Auto-generated catch block

+ 27 - 10
collect/src/main/java/com/saga/test/Sunyifu.java

@@ -1,6 +1,5 @@
 package com.saga.test;
 
-import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileOutputStream;
 import java.io.OutputStream;
@@ -11,7 +10,6 @@ import java.util.Map;
 import java.util.Map.Entry;
 
 import org.apache.hadoop.conf.Configuration;
-import org.apache.poi.hssf.usermodel.HSSFCell;
 import org.apache.poi.hssf.usermodel.HSSFRow;
 import org.apache.poi.hssf.usermodel.HSSFSheet;
 import org.apache.poi.hssf.usermodel.HSSFWorkbook;
@@ -47,12 +45,32 @@ public class Sunyifu {
 			// Thread.sleep(1L);
 			// }
 			// Thread.sleep(Constant.Millisecond.MINUTE_30);
-			LogUtil.info("get data begin...");
 			String timefrom = "20170701000000";
-			String timeto = "20170801000000";
+			String timeto = "20180901000000";
 			// long[] funcs = { 11011, 11012, 11013, 11014 };
-
 			GetPointList(agent, "db_time_data", "dy_pointlist", "1101080002");
+
+			Calendar calendarFrom = Calendar.getInstance();
+			Calendar calendarTo = Calendar.getInstance();
+			Date from = Constant.parse(timefrom);
+			Date to = Constant.parse(timeto);
+			calendarFrom.setTime(from);
+			calendarTo.setTime(to);
+			while (calendarFrom.before(calendarTo)) {
+				Date datetmp = calendarFrom.getTime();
+				calendarFrom.add(Calendar.MONTH, 1);
+
+				export2(Constant.format(datetmp), Constant.format(calendarFrom.getTime()));
+			}
+		} catch (Exception e) {
+			// TODO Auto-generated catch block
+			LogUtil.error("Main: " + e.toString());
+		}
+	}
+
+	private static void export2(String timefrom, String timeto) {
+		try {
+			LogUtil.info("get data begin...");
 			for (Entry<String, String> m : meterList.entrySet()) {
 				dataList.put(m.getKey(), new HashMap<String, Map<String, Double>>());
 				for (long funcid : funcs) {
@@ -76,7 +94,6 @@ public class Sunyifu {
 			calendarFrom.setTime(from);
 			calendarTo.setTime(to);
 
-			// 模拟数据
 			// 创建HSSFWorkbook对象(excel的文档对象)
 			HSSFWorkbook wb = new HSSFWorkbook();
 			for (long funcid : funcs) {
@@ -113,13 +130,13 @@ public class Sunyifu {
 			}
 
 			// 输出Excel文件
-			OutputStream output = new FileOutputStream("d:/data.xls");
+			OutputStream output = new FileOutputStream("d:/data" + timefrom.substring(0, 6) + ".xls");
 			wb.write(output);
 			output.close();
 			LogUtil.info("insert excel end...");
-		} catch (Exception e) {
+		} catch (Exception e1) {
 			// TODO Auto-generated catch block
-			LogUtil.error("Main: " + e.toString());
+			e1.printStackTrace();
 		}
 	}
 
@@ -151,7 +168,7 @@ public class Sunyifu {
 
 			Double data = (Double) dataItem.get("data_value");
 			dataList.get(meter).get(String.valueOf(funcid)).put(receivetime, data);
-			LogUtil.info(("receivetime " + receivetime + " data:" + data));
+			// LogUtil.info(("receivetime " + receivetime + " data:" + data));
 		}
 	}
 

+ 23 - 3
collect/src/main/java/com/saga/thread/down/CollectThread.java

@@ -93,8 +93,18 @@ public class CollectThread extends Thread {
 						if (address != null) {
 							for (Entry<String, IServerManager> server : this.serverList.entrySet()) {
 								try {
-									String cmd = StringUtils.join(ps.content, ";").replace("pointset", "senddownset")
-											.replace(ps.buildingSignUP, ps.buildingSign);
+									// String cmd = StringUtils.join(ps.content,
+									// ";").replace("pointset", "senddownset")
+									// .replace(ps.buildingSignUP,
+									// ps.buildingSign);
+									// 1101070037;1;senddownset;;123;1001;11;3.1
+									StringBuffer sb = new StringBuffer();
+									
+									sb.append(ps.buildingSign).append(";").append(ps.mac).append(";senddownset;;")
+											.append(ps.ID).append(";").append(ps.meterSign).append(";")
+											.append(ps.funcID).append(";").append(ps.dataSet).append(";");
+									
+									String cmd = sb.toString();
 
 									String id = ps.buildingSignUP + "-" + ps.mac + "-" + ps.ID;
 									Constant.id2status.put(id, "processing:sent");
@@ -150,6 +160,8 @@ public class CollectThread extends Thread {
 					// p.address + " " + s);
 					String[] rList = s.split(";");
 					if ("report".equalsIgnoreCase(rList[2]) && rList.length > 7) {
+						// 1101070037;1;report;20010203040506;123;1001;2;11;3.1;12;3.2
+						Constant.pointMAC.put(rList[0] + "-" + rList[5], rList[1]);
 						Constant.collectCount++;
 						Constant.nodeList.put(rList[0] + "-" + rList[1], System.currentTimeMillis());
 						// LogUtil.info("CollectThread Challenge
@@ -199,6 +211,7 @@ public class CollectThread extends Thread {
 						}
 						// }
 					} else if ("reportaddtion".equalsIgnoreCase(rList[2]) && rList.length > 7) {
+						Constant.pointMAC.put(rList[0] + "-" + rList[5], rList[1]);
 						Constant.collectCount++;
 						Constant.nodeList.put(rList[0] + "-" + rList[1], System.currentTimeMillis());
 						// LogUtil.info("CollectThread Challenge
@@ -256,8 +269,15 @@ public class CollectThread extends Thread {
 							Constant.id2status.put(id, "processing:finish:" + rList[7]);
 						}
 						this.updatePointSet(rList);
+						// 1101070037;1;senddownsetack;20010203040506;123;1001;11;success
+						String MAC = rList[1];
+
+						if (Constant.pointMAC.containsKey(rList[0] + "-" + rList[5])) {
+							MAC = Constant.pointMAC.get(rList[0] + "-" + rList[5]);
+							LogUtil.info("MAC:" + (rList[0] + "-" + rList[5]) + " " + rList[1] + "-->" + MAC);
+						}
 						HBaseUtil.InsertPointCommunicationLog(Constant.agent, Constant.BuildingDB.get(rList[0]),
-								rList[0], this.ip, rList[1], rList[5], (rList[6]), "down", "senddownsetack", rList[4],
+								rList[0], this.ip, MAC, rList[5], (rList[6]), "down", "senddownsetack", rList[4],
 								Constant.format(new Date()), "processing:finish:" + rList[7]);
 
 						// LogUtil.info("id: " + id);

+ 43 - 22
collect/src/main/java/com/saga/thread/up/SendThread.java

@@ -2,7 +2,9 @@ package com.saga.thread.up;
 
 import java.net.UnknownHostException;
 import java.util.Date;
+import java.util.HashMap;
 import java.util.Iterator;
+import java.util.Map;
 
 import org.zillion.util.log.LogUtil;
 
@@ -18,6 +20,7 @@ public class SendThread extends Thread {
 
 	private IServerManager uploadServer = null;
 	private String ip = null;
+	private Map<String, String> pointMac_UP = new HashMap<String, String>();
 
 	public SendThread(IServerManager uploadServer) {
 		// TODO Auto-generated constructor stub
@@ -78,7 +81,8 @@ public class SendThread extends Thread {
 						StringBuffer sb = new StringBuffer();
 						PointSet ps = it.next();
 						// String address =
-						// Constant.CollectorPointSetList.get(ps.buildingSign + ";"
+						// Constant.CollectorPointSetList.get(ps.buildingSign +
+						// ";"
 						// + ps.ID + ";");
 						// String address = ps.address;
 
@@ -89,23 +93,27 @@ public class SendThread extends Thread {
 						if (ps.address == null) {
 							break;
 						}
-
+						String MAC = ps.mac;
+						if (this.pointMac_UP.containsKey(ps.buildingSignUP + "-" + ps.meterSign)) {
+							MAC = this.pointMac_UP.get(ps.buildingSignUP + "-" + ps.meterSign);
+						}
 						// 已得到控制设备的反馈状态,移除
 						if (ps.recTime > 0 && ps.sendTime > 0 && (ps.status != null && ps.status.length() > 0)) {
-							sb.append(ps.buildingSignUP).append(";").append(ps.mac).append(";").append("pointsetack")
-									.append(";").append(Constant.format(new Date())).append(";").append(ps.ID).append(";")
-									.append(ps.meterSign).append(";").append(ps.funcID).append(";").append(ps.status)
-									.append(";");
+							sb.append(ps.buildingSignUP).append(";").append(MAC).append(";").append("pointsetack")
+									.append(";").append(Constant.format(new Date())).append(";").append(ps.ID)
+									.append(";").append(ps.meterSign).append(";").append(ps.funcID).append(";")
+									.append(ps.status).append(";");
 							this.uploadServer.AppendToSend(ps.address, new Packet(sb.toString()));
 							Constant.sendSend += sb.toString().length();
 							HBaseUtil.BatchInsert_Set2(Constant.agent, Constant.BuildingDB.get(ps.buildingSign), ps);
-							LogUtil.info(this.uploadServer.Name() + " AppendToSend: " + ps.address + " " + sb.toString());
+							LogUtil.info(
+									this.uploadServer.Name() + " AppendToSend: " + ps.address + " " + sb.toString());
 
 							String id = ps.buildingSignUP + "-" + ps.mac + "-" + ps.ID;
 							HBaseUtil.InsertPointCommunicationLog(Constant.agent,
 									Constant.BuildingDB.get(ps.buildingSignUP), ps.buildingSignUP, this.ip, ps.mac,
-									ps.meterSign, String.valueOf(ps.funcID), "down", "pointsetack", String.valueOf(ps.ID),
-									Constant.format(new Date()), "finish:finish:" + ps.status);
+									ps.meterSign, String.valueOf(ps.funcID), "down", "pointsetack",
+									String.valueOf(ps.ID), Constant.format(new Date()), "finish:finish:" + ps.status);
 							synchronized (Constant.id2status) {
 								Constant.id2status.put(id, "finish:finish:" + ps.status);
 							}
@@ -123,27 +131,30 @@ public class SendThread extends Thread {
 						} // 超时未得到设备的反馈状态,移除
 						else if (ps.recTime == 0 && ps.sendTime > 0
 								&& (System.currentTimeMillis() - ps.sendTime > Constant.getWaitingtime())) {
-							sb.append(ps.buildingSignUP).append(";").append(ps.mac).append(";").append("pointsetack")
-									.append(";").append(Constant.format(new Date())).append(";").append(ps.ID).append(";")
-									.append(ps.meterSign).append(";").append(ps.funcID).append(";").append("fail:timeout")
-									.append(";");
+							sb.append(ps.buildingSignUP).append(";").append(MAC).append(";").append("pointsetack")
+									.append(";").append(Constant.format(new Date())).append(";").append(ps.ID)
+									.append(";").append(ps.meterSign).append(";").append(ps.funcID).append(";")
+									.append("fail:timeout").append(";");
 							this.uploadServer.AppendToSend(ps.address, new Packet(sb.toString()));
 							Constant.sendSend += sb.toString().length();
 							HBaseUtil.BatchInsert_Set3(Constant.agent, Constant.BuildingDB.get(ps.buildingSign), ps);
-							LogUtil.info(this.uploadServer.Name() + " AppendToSend: " + ps.address + " " + sb.toString());
+							LogUtil.info(
+									this.uploadServer.Name() + " AppendToSend: " + ps.address + " " + sb.toString());
 							String id = ps.buildingSignUP + "-" + ps.mac + "-" + ps.ID;
 
 							// LogUtil.info("InsertPointCommunicationLog" + "
 							// begin");
 							HBaseUtil.InsertPointCommunicationLog(Constant.agent,
 									Constant.BuildingDB.get(ps.buildingSignUP), ps.buildingSignUP, this.ip, ps.mac,
-									ps.meterSign, String.valueOf(ps.funcID), "down", "pointsetack", String.valueOf(ps.ID),
-									Constant.format(new Date()), "finish:finish:" + "fail:timeout");
+									ps.meterSign, String.valueOf(ps.funcID), "down", "pointsetack",
+									String.valueOf(ps.ID), Constant.format(new Date()),
+									"finish:finish:" + "fail:timeout");
 
 							synchronized (Constant.id2status) {
 								Constant.id2status.put(id, "finish:finish:" + "fail:timeout");
 							}
-							// LogUtil.info("InsertPointCommunicationLog" + " end");
+							// LogUtil.info("InsertPointCommunicationLog" + "
+							// end");
 
 							// CSVUtil.insert(Constant.setlog,
 							// new StringBuffer(
@@ -188,9 +199,14 @@ public class SendThread extends Thread {
 			for (String result : resultArray) {
 				String[] rList = result.split(";");
 				if (rList.length > 7) {
-					String id = rList[0] + "-" + rList[1] + "-" + rList[4];
-					String key = rList[0] + "-" + rList[1] + "-" + rList[5] + "." + rList[6];
+					String MAC = rList[1];
 
+					if (Constant.pointMAC.containsKey(rList[0] + "-" + rList[5])) {
+						MAC = Constant.pointMAC.get(rList[0] + "-" + rList[5]);
+						LogUtil.info("MAC:" + (rList[0] + "-" + rList[5]) + " " + rList[1] + "-->" + MAC);
+					}
+					String id = rList[0] + "-" + MAC + "-" + rList[4];
+					String key = rList[0] + "-" + MAC + "-" + rList[5] + "." + rList[6];
 					// Constant.id2time.put(id, System.currentTimeMillis());
 					// Constant.id2point.put(id, key);
 					// Constant.id2function.put(id, rList[2]);
@@ -198,7 +214,7 @@ public class SendThread extends Thread {
 
 					try {
 						HBaseUtil.InsertPointCommunicationLog(Constant.agent, Constant.BuildingDB.get(rList[0]),
-								rList[0], this.ip, rList[1], rList[5], (rList[6]), this.getupdown(rList[2]), rList[2],
+								rList[0], this.ip, MAC, rList[5], (rList[6]), this.getupdown(rList[2]), rList[2],
 								String.valueOf(rList[4]), Constant.format(new Date()), "processing:rcvd");
 					} catch (UnknownHostException e1) {
 						// TODO Auto-generated catch block
@@ -231,12 +247,17 @@ public class SendThread extends Thread {
 						ps.recTime = 0;
 						ps.buildingSign = rList[0];
 						ps.buildingSignUP = rList[0];
-						if (Constant.BuildingConv.containsKey(ps.buildingSign)) {
+						//楼号有转换规则,且转换后的楼-表有数据才进行转换
+						//解决一楼多楼号时总有部分仪表无法下发的问题
+						if (Constant.BuildingConv.containsKey(ps.buildingSign) && Constant.Meter_Collector_Control
+								.containsKey(Constant.BuildingConv.get(ps.buildingSign) + "-" + ps.meterSign)) {
 							ps.buildingSign = Constant.BuildingConv.get(ps.buildingSign);
 						}
-						ps.mac = rList[1];
+
 						ps.ID = Integer.valueOf(rList[4]);
 						ps.meterSign = rList[5];
+						ps.mac = MAC;
+						this.pointMac_UP.put(ps.buildingSignUP + "-" + ps.meterSign, rList[1]);
 						ps.funcID = Integer.valueOf(rList[6]);
 						ps.dataSet = Double.valueOf(rList[7]);
 						// String adress = packageRece.address;

+ 1 - 0
collect/src/main/java/com/saga/util/Constant.java

@@ -68,6 +68,7 @@ public class Constant {
 	public static int collectCount = 0;
 	public static int collectReportCount = 0;
 	public static int report2DBCount = 0;
+	public static Map<String, String> pointMAC = new HashMap<String, String>();
 
 	static {
 		String resource = "/config.json";