Merge branch 'master'

conf/gateway.conf

@@ -226,5 +226,9 @@
 	"pro_category":1,
 	"syn_data":false,
 	"syn_extension_Time":false,
-	"record_log_path":"/usr/local/dpark"
+	"record_log_path":"/usr/local/dpark",
+	"mqtt_open":true,
+	"mqtt_broker":"tcp://120.77.206.101:1883",
+	"mqtt_username":"",
+	"mqtt_password":""
 }

loadconfig.go

@@ -9,6 +9,7 @@ import (
 	// "strconv"
 	"time"
 	"sync"
+	"math/rand"
 )
 
 type Traffic struct {
@@ -43,6 +44,10 @@ type Conf struct {
 	SYNDATA				bool `json:"syn_data"`
 	SYNEXTENSIONTIME bool `json:"syn_extension_Time"`
 	RECORD_LOG_PATH string `json:"record_log_path"`
+	MQTT_OPEN bool `json:"mqtt_open"`
+	MQTT_BROKER string `json:"mqtt_broker"`
+	MQTT_USERNAME string `json:"mqtt_username"`
+	MQTT_PASSWORD string `json:"mqtt_password"`
 }
 
 //无线 下位机
@@ -94,6 +99,7 @@ type LoraNode struct{
 	CURRENT_DATA		int 		`json:"current_data"`
 	CHECK_TIME			bool 		`json:"check_time"`
 	DEVICE 					string	`json:"device"`
+	UPDATE_TIME				time.Time `json:"update_time"`
 }
 
 var conf Conf
@@ -101,13 +107,11 @@ var DEVICEDATAS Devicedatas
 var demoDevice Device
 
 var today string
-//var airDevice Device
-
-//var femaledemoDevice Device
-//var maledemoDevice Device
 
 var nodeUpdateMap sync.Map
-// var logch = make(chan []byte)
+
+//sensor data update map
+var sensorUpdateMap sync.Map
 
 //log channel
 var logch = make(chan []byte)
@@ -120,6 +124,9 @@ var db, _ = sql.Open("sqlite3", "./db/gateway.db")
 // 加载配置文件
 func init() {
 
+	//以时间配置随机数种子
+	rand.Seed(time.Now().UnixNano())
+
 	t := time.Now()
  	today = t.Format("2006-01-02")
 

main.go

@@ -52,6 +52,7 @@ func main() {
 	c.AddFunc("0, */10, *, *, *, *", savedata_cron)
 	c.AddFunc("0, 0, 0, *, *, *", insertdata)
 	c.AddFunc("0, 0, *, *, *, *", get_weather)
+	//定时更新today字段，当前日期
 	c.AddFunc("0, *, *, *, *, *", func(){
 		t := time.Now()
  		today = t.Format("2006-01-02")
@@ -60,6 +61,8 @@ func main() {
 
 	if conf.OPENTCPSETVER {
 
+		//开启mqtt默认认定是无线方案
+		
 		go record_log_channel()
 
 		tcp_port := strconv.Itoa(conf.TCPPORT)

router.go

@@ -232,7 +232,7 @@ func dealDatav3(data []byte) string{
 		// log.Println("loraNodeMap:",loraNodeMap)
 		loraNode, ok := loraNodeMap.Load(adr)
 		if !ok {
-			log.Println("load loraNode success ", ok)
+			log.Println("load loraNode fail ", adr)
 			return ""
 		}
 		// log.Println("load loraNode:",loraNode)
@@ -362,6 +362,8 @@ func dealDatav3(data []byte) string{
 		// log.Println("data[9]",int64(data[9]))
 		// log.Println("data[10]",int64(data[10]))
 
+		log.Println("aqdata|all:",data)
+
 		if data[6] == 0{
 			buffer.WriteString("-")
 		}
@@ -376,10 +378,10 @@ func dealDatav3(data []byte) string{
 		}
 		buffer.WriteString(wdint2)
 
-		log.Println("wdfh",data[6])
+		log.Println("aqdata|wdfh",data[6])
 
 		wddata := buffer.String()
-		log.Println("wddata",wddata)
+		log.Println("aqdata|wddata",wddata)
 
 		buffer = new(bytes.Buffer)
 
@@ -393,7 +395,7 @@ func dealDatav3(data []byte) string{
 		buffer.WriteString(sdint2)
 
 		sddata := buffer.String()
-		log.Println("sddata",sddata)
+		log.Println("aqdata|sddata",sddata)
 
 		buffer = new(bytes.Buffer)
 
@@ -407,38 +409,50 @@ func dealDatav3(data []byte) string{
 		buffer.WriteString(nh3int2)
 		nh3data := buffer.String()
 
-		log.Println("nh3",nh3data)
-		log.Println("h2s",data[13])
+		log.Println("aqdata|nh3",nh3data)
+
+		
+		buffer = new(bytes.Buffer)
+		h2sint1 := strconv.Itoa(int(data[13]))
+		buffer.WriteString(h2sint1)
+		buffer.WriteString(".")
+		h2sint2 := strconv.Itoa(int(data[14]))
+		if int64(data[14])<10{
+			buffer.WriteString("0")
+		}
+		buffer.WriteString(h2sint2)
+		h2sdata := buffer.String()
+		log.Println("aqdata|h2s",h2sdata)
 
 		buffer = new(bytes.Buffer)
-		ch2o1 := strconv.Itoa(int(data[14]))
+		ch2o1 := strconv.Itoa(int(data[15]))
 		buffer.WriteString(ch2o1)
 		buffer.WriteString(".")
-		ch2o2 := strconv.Itoa(int(data[15]))
-		if int64(data[15])<10{
+		ch2o2 := strconv.Itoa(int(data[16]))
+		if int64(data[16])<10{
 			buffer.WriteString("0")
 		}
 		buffer.WriteString(ch2o2)
 		ch2o := buffer.String()
 
 		// ch2o := uint32(data[14])*256+uint32(data[15])
-		log.Println("ch2o",ch2o)
+		log.Println("aqdata|ch2o",ch2o)
 
-		co2data := uint32(data[16])*256+uint32(data[17])
-		log.Println("co2data",co2data)
+		co2data := uint32(data[17])*256+uint32(data[18])
+		log.Println("aqdata|co2data",co2data)
 
-		tvocdata := data[18]
-		log.Println("tvocdata",tvocdata)
+		tvocdata := data[19]
+		log.Println("aqdata|tvocdata",tvocdata)
 
 		// log.Println("pm25data1",data[19])
 		// log.Println("pm25data2",data[20])
-		pm25data := uint32(data[19])*256+uint32(data[20])
-		log.Println("pm25data",pm25data)
+		pm25data := uint32(data[20])*256+uint32(data[21])
+		log.Println("aqdata|pm25data",pm25data)
 
 		// log.Println("pm10data1",data[21])
 		// log.Println("pm10data2",data[22])
-		pm10data := uint32(data[21])*256+uint32(data[22])
-		log.Println("pm10data",pm10data)
+		pm10data := uint32(data[22])*256+uint32(data[23])
+		log.Println("aqdata|pm10data",pm10data)
 
 		var R_data P1004_6
 		R_data.CODE = 1005
@@ -452,7 +466,7 @@ func dealDatav3(data []byte) string{
 		R_data.DATA["humidity"]=sddata
 
 		R_data.DATA["nh3"]=nh3data
-		R_data.DATA["h2s"]=data[13]
+		R_data.DATA["h2s"]=h2sdata
 		R_data.DATA["ch2o"]=ch2o
 		R_data.DATA["co2"]=co2data
 		R_data.DATA["tvoc"]=tvocdata

tcpserver.go

@@ -85,6 +85,7 @@ func readConn(conn net.Conn, readChan chan<- []byte, stopChan chan<- bool) {
 		continueCount := 0
 		// log.Println("yunhorndeug.push.data.tcp.reqLen|data",reqLen,data)
 		log.Println("$$$$$$$$$$$$$$$$$$$$$$$$$",data[:reqLen])
+		now := time.Now()
     for i, b := range data[:reqLen] {
 
 				if reqLen<7{
@@ -104,11 +105,6 @@ func readConn(conn net.Conn, readChan chan<- []byte, stopChan chan<- bool) {
 	        buffer.WriteString(s)
 				}
 
-				// log.Println("========================",data[reqLen-2:reqLen])
-
-				// log.Println(b)
-				// log.Println("$$$$$$$$$$$$$$$$$$$$$$$$$",i,reqLen)
-				// log.Println("$$$$$$$$$$$$$$$$$$$$$$$$$",data[reqLen])
 				end :=0
 				//254为分隔符
 				if b==254 && i>3{
@@ -150,8 +146,8 @@ func readConn(conn net.Conn, readChan chan<- []byte, stopChan chan<- bool) {
 						values = data[i+2:i+3]
 					}else if dataType ==4 {
 						//空气质量
-						continueCount = 18
-						end = 19
+						continueCount = 19
+						end = 20
 						if i+end >reqLen{
 							log.Println("yunhorndeug.push.data.tcp.fq",buffer.String(),addrs,dataType)
 							continue
@@ -215,13 +211,23 @@ func readConn(conn net.Conn, readChan chan<- []byte, stopChan chan<- bool) {
 
 					log.Println("yunhorndeug.push.data.tcp:",buffer.String(),addrs,dataType,values)
 
-					_,ok := loraNodeMap.Load(buffer.String())
+					loraNode,ok := loraNodeMap.Load(buffer.String())
 					if !ok{
-						log.Println("yunhorndeug.push.data.tcp.dev not exist:",buffer.String())
-							buffer = new(bytes.Buffer)
+						log.Println("yunhorndeug.push.data.tcp.dev not exist:",buffer.String(),dataType,values)
+						buffer = new(bytes.Buffer)
 						continue
 					}
 
+					if !loraNode.(LoraNode).UPDATE_TIME.IsZero() {
+						var duration = now.Sub(loraNode.(LoraNode).UPDATE_TIME).Seconds()
+						var durationAsInt64 = int(duration)
+						if(durationAsInt64<=1){
+							log.Println("repeat.sensor.data:",addrs)
+							continue;
+						}
+					}
+					
+
 						buffer = new(bytes.Buffer)
 
 						go write_log_channel(data[i-4:i+end])