diff --git a/conf/gateway.conf b/conf/gateway.conf
index 4b35738..f65dd14 100644
--- a/conf/gateway.conf
+++ b/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":""
}
diff --git a/loadconfig.go b/loadconfig.go
index 31c47ad..e58f5d3 100644
--- a/loadconfig.go
+++ b/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")
diff --git a/main.go b/main.go
index e35baad..7b4f00b 100644
--- a/main.go
+++ b/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)
diff --git a/router.go b/router.go
index 4892a51..b0db176 100644
--- a/router.go
+++ b/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
diff --git a/tcpserver.go b/tcpserver.go
index db779c7..4c2a940 100644
--- a/tcpserver.go
+++ b/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])