增加SetOffset

config/config.toml

@@ -1,6 +1,6 @@
 [Server]
     Addr = "0.0.0.0"
-    Port = 8999
+    Port = 8997
     RunMode = "debug"
 [Sparrow]
     Server = "http://192.168.0.224:18100"

go.mod

@@ -17,4 +17,4 @@ require (
 	golang.org/x/text v0.3.8-0.20211105212822-18b340fc7af2 // indirect
 )
 
-replace sparrow-sdk v1.0.0 => gogs.yehaoji.cn/yongxu/sparrow-sdk.git v1.1.2
+replace sparrow-sdk v1.0.0 => gogs.yehaoji.cn/yongxu/sparrow-sdk.git v1.1.4

go.sum

@@ -61,8 +61,8 @@ go.opentelemetry.io/otel/trace v1.0.0-RC2/go.mod h1:JPQ+z6nNw9mqEGT8o3eoPTdnNI+A
 go.opentelemetry.io/otel/trace v1.0.0/go.mod h1:PXTWqayeFUlJV1YDNhsJYB184+IvAH814St6o6ajzIs=
 go.opentelemetry.io/otel/trace v1.7.0 h1:O37Iogk1lEkMRXewVtZ1BBTVn5JEp8GrJvP92bJqC6o=
 go.opentelemetry.io/otel/trace v1.7.0/go.mod h1:fzLSB9nqR2eXzxPXb2JW9IKE+ScyXA48yyE4TNvoHqU=
-gogs.yehaoji.cn/yongxu/sparrow-sdk.git v1.1.2 h1:Iej//+HmxZSKIwP+1sbSAegE7KErswmsBwCEQLZ17jM=
-gogs.yehaoji.cn/yongxu/sparrow-sdk.git v1.1.2/go.mod h1:hWw7D5hrW8f8cOKKdhtlt8HQbdfD2o6PllWMhs0BdQs=
+gogs.yehaoji.cn/yongxu/sparrow-sdk.git v1.1.4 h1:7DEfCFizL5wTNe/0cD8SMNR0W2TFEGVTyeUProeaaEc=
+gogs.yehaoji.cn/yongxu/sparrow-sdk.git v1.1.4/go.mod h1:hWw7D5hrW8f8cOKKdhtlt8HQbdfD2o6PllWMhs0BdQs=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=

main.go

@@ -1,13 +1,16 @@
 package main
 
 import (
+	"bth-rs30-gateway/protocol"
 	"bth-rs30-gateway/server"
 	"context"
+	"github.com/gogf/gf/encoding/gjson"
 	"github.com/gogf/gf/frame/g"
 	"github.com/gogf/gf/os/glog"
 	"github.com/gogf/gf/os/gproc"
 	"os"
 	"sparrow-sdk/config"
+	"sparrow-sdk/protocal"
 	gatewayV2 "sparrow-sdk/v2"
 )
 
@@ -48,6 +51,39 @@ func main() {
 			panic(err)
 		}
 	}()
+	closeReportChan := make(chan struct{})
+	go func() {
+		for {
+			select {
+			case msg := <-gw.RecvCommand():
+				if msg.Data.Cmd == "SetOffset" {
+
+				}
+			case <-closeReportChan:
+				return
+			}
+		}
+	}()
+	if err = gw.RegisterCommand("SetOffset", func(msg protocal.CloudSend) error {
+		var params protocol.SetOffSetParams
+		j := gjson.New(msg.Data.Params)
+		err = j.Struct(&params)
+		if err != nil {
+			glog.Errorf("错误的指令参数%s", err.Error())
+			return err
+		}
+		glog.Debugf("指令:%s, 子设备Id:%s, 参数:%v", msg.Data.Cmd, msg.SubDeviceId, params)
+		client := srv.GetClient(msg.SubDeviceId)
+		if client != nil {
+			if err = client.SetOffset(params.Act, params.Value, 1); err != nil {
+				glog.Errorf("执行命令出错:%s", err.Error())
+				return err
+			}
+		}
+		return nil
+	}); err != nil {
+		panic(err)
+	}
 
 	gproc.AddSigHandlerShutdown(func(sig os.Signal) {
 		gw.Close()

protocol/protocol.go

@@ -4,3 +4,8 @@ type Data struct {
 	Temperature float32 `json:"temperature"` // 温度
 	Humidly     float32 `json:"humidly"`     // 湿度
 }
+
+type SetOffSetParams struct {
+	Act   int `json:"act"`   // 操作类型：(1:使能温度校准2:关闭温度校准 3:使能湿度校准 4:关闭湿度校准）
+	Value int `json:"value"` // 偏移值
+}

server/client.go

@@ -2,14 +2,17 @@ package server
 
 import (
 	"bth-rs30-gateway/protocol"
+	"bytes"
+	"encoding/binary"
 	"errors"
 	"fmt"
 	"github.com/gogf/gf/encoding/gbinary"
-	"github.com/gogf/gf/frame/g"
 	"github.com/gogf/gf/net/gtcp"
 	"github.com/gogf/gf/os/glog"
+	"github.com/gogf/gf/util/gconv"
 	"io"
 	"net"
+	"strconv"
 	"strings"
 	"syscall"
 	"time"
@@ -22,6 +25,7 @@ type Client struct {
 	sendChan     chan []byte
 	closeChan    chan struct{}
 	closeHandler func(id string, c *Client)
+	regHandler   func(id string, c *Client)
 	isReg        bool
 }
 
@@ -55,16 +59,20 @@ func (c *Client) SendLoop() {
 					glog.Error("接收指令超时")
 					break
 				default:
+
+					receiveBuf, err := c.conn.Recv(-1)
+					if err != nil {
+						c.readError(err)
+						break
+					}
 					if !c.isReg {
-						id := gbinary.DecodeToString(buf)
+						id := gbinary.DecodeToString(receiveBuf)
 						glog.Debugf("收到注册包!id:%s", id)
 						c.SetId(id)
 						c.isReg = true
-						break
-					}
-					receiveBuf, err := c.conn.Recv(-1)
-					if err != nil {
-						c.readError(err)
+						if c.regHandler != nil {
+							c.regHandler(c.Id, c)
+						}
 						break
 					}
 					glog.Debugf("收到数据：%2X", receiveBuf)
@@ -89,10 +97,10 @@ func (c *Client) decodeAndReport(buf []byte) error {
 		return errors.New(fmt.Sprintf("modbus: response crc '%v' does not match expected '%v'", checksum, crc.value()))
 	}
 
-	if buf[1] == 0x03 && buf[2] == 0x06 {
+	if buf[1] == 0x03 {
 		data := &protocol.Data{}
-		data.Temperature = float32(gbinary.BeDecodeToUint16(buf[3:5])) * 0.1
-		data.Humidly = float32(gbinary.BeDecodeToUint16(buf[5:7])) * 0.1
+		data.Temperature = gconv.Float32(caleTemperature(buf[3:5])) * 0.1
+		data.Humidly = gconv.Float32(gbinary.BeDecodeToUint16(buf[5:7])) * 0.1
 		if err := c.srv.ReportStatus(c.Id, data); err != nil {
 			return err
 		}
@@ -117,6 +125,28 @@ func (c *Client) closeConnection() {
 	}
 }
 
+// 计算温度值, 处理零下的情况
+func caleTemperature(data []byte) int {
+	var ym uint16
+	var isBlowZero bool
+	var result int
+	bm := binary.BigEndian.Uint16(data)
+	var bitNum = len(data) * 8
+	f := "%." + strconv.Itoa(bitNum) + "b"
+	bmStr := fmt.Sprintf(f, bm)
+	if string(bmStr[0]) == "1" { // blow zero
+		ym = ^bm + 1
+		isBlowZero = true
+	} else {
+		ym = bm
+	}
+	result = int(ym)
+	if isBlowZero {
+		result = int(ym) * -1
+	}
+	return result
+}
+
 // isErrConnReset read: connection reset by peer
 func isErrConnReset(err error) bool {
 	if ne, ok := err.(*net.OpError); ok {
@@ -125,10 +155,55 @@ func isErrConnReset(err error) bool {
 	return false
 }
 
+// SetOffset 设置温湿度偏移值
+func (c *Client) SetOffset(act, value, slaveId int) error {
+	var sendBuf []byte
+	buffer := bytes.NewBuffer(sendBuf)
+	buffer.Write(gbinary.BeEncodeInt(slaveId))
+
+	var funcByte byte
+	var address []byte
+	if act == 1 || act == 3 {
+		funcByte = 0x10
+		address = []byte{0x00, 0x50}
+	}
+	if act == 2 || act == 4 {
+		funcByte = 0x06
+		address = []byte{0x00, 0x52}
+	}
+	buffer.WriteByte(funcByte)
+	buffer.Write(address)
+
+	if act == 1 || act == 3 {
+		number := []byte{0x00, 0x02}
+		buffer.Write(number)
+		buffer.Write([]byte{0x00, 0x01})
+		buffer.Write(dataBlock(value))
+	} else if act == 2 || act == 4 {
+		buffer.Write([]byte{0x00, 0x00})
+	}
+
+	var mCrc crc
+	checkSum := mCrc.reset().pushBytes(buffer.Bytes()).value()
+	buffer.Write([]byte{byte(checkSum), byte(checkSum >> 8)})
+	return c.send(buffer.Bytes())
+}
+
+func dataBlock(value int) []byte {
+	buffer := &bytes.Buffer{}
+	if value < 0 {
+		buffer.WriteByte(0xFF)
+	} else {
+		buffer.WriteByte(0x00)
+	}
+	buffer.Write(gbinary.BeEncodeInt(value))
+	return buffer.Bytes()
+}
+
 func (c *Client) GetSendByte() {
 	for {
 		c.sendChan <- []byte{0x01, 0x03, 0x00, 0x00, 0x00, 0x02, 0xC4, 0x0B}
-		time.Sleep(time.Duration(g.Cfg().GetInt("Server.Frequency")) * time.Second)
+		time.Sleep(10 * time.Second)
 	}
 }
 

server/server.go

@@ -3,9 +3,12 @@ package server
 import (
 	"context"
 	"fmt"
+	"github.com/gogf/gf/container/gmap"
+	"github.com/gogf/gf/frame/g"
 	"github.com/gogf/gf/net/gtcp"
 	"github.com/gogf/gf/os/glog"
 	gatewayV2 "sparrow-sdk/v2"
+	"time"
 )
 
 type Server struct {
@@ -15,6 +18,7 @@ type Server struct {
 	addr      string
 	port      int
 	gateWay   *gatewayV2.Gateway
+	clients   *gmap.Map
 }
 
 func NewServer(ctx context.Context, addr string, port int, gw *gatewayV2.Gateway) *Server {
@@ -24,6 +28,7 @@ func NewServer(ctx context.Context, addr string, port int, gw *gatewayV2.Gateway
 		addr:      addr,
 		port:      port,
 		gateWay:   gw,
+		clients:   gmap.New(false),
 	}
 }
 
@@ -35,19 +40,40 @@ func (s *Server) Start() error {
 }
 
 func (s *Server) Stop() {
-	s.srv.Close()
+	s.clients.Iterator(func(k interface{}, v interface{}) bool {
+		client := v.(*Client)
+		close(client.closeChan)
+		return true
+	})
+	_ = s.srv.Close()
 }
 
 func (s *Server) onClientConnect(conn *gtcp.Conn) {
-	glog.Debugf("新的设备接入:%s", conn.RemoteAddr())
 	client := NewClient(s, conn)
 	client.closeHandler = func(id string, c *Client) {
 		glog.Debugf("客户端断开:%s", id)
+		if id != "" {
+			_ = s.gateWay.SubDeviceLogout(g.Cfg().GetString("sparrow.DeviceCode"), id)
+			s.clients.Remove(id)
+		}
+	}
+	client.regHandler = func(id string, c *Client) {
+		_ = s.gateWay.SubDeviceLogin(g.Cfg().GetString("Sparrow.DeviceCode"), id)
+		s.clients.Set(id, c)
 	}
 	go client.SendLoop()
+	time.Sleep(10 * time.Second)
 	go client.GetSendByte()
 }
 
 func (s *Server) ReportStatus(subId string, data interface{}) error {
 	return s.gateWay.ReportStatus(subId, "status", data)
 }
+
+func (s *Server) GetClient(subId string) *Client {
+	client := s.clients.Get(subId)
+	if client != nil {
+		return client.(*Client)
+	}
+	return nil
+}

vendor/modules.txt

@@ -151,7 +151,7 @@ golang.org/x/text/runes
 golang.org/x/text/transform
 # gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b
 gopkg.in/yaml.v3
-# sparrow-sdk v1.0.0 => gogs.yehaoji.cn/yongxu/sparrow-sdk.git v1.1.2
+# sparrow-sdk v1.0.0 => gogs.yehaoji.cn/yongxu/sparrow-sdk.git v1.1.4
 ## explicit
 sparrow-sdk/config
 sparrow-sdk/errors

vendor/sparrow-sdk/config/config.go

@@ -13,4 +13,7 @@ type Config struct {
 	Version       string            // 网关的版本号
 	Debug         bool              //是否显示调试信息
 	Logger        logger.Interface  // 日志组件
+	UseTls        bool              // 是否通过ssl接入平台，如果启用，要配置CaFile 和kKeyFile两个参数
+	CaFile        string            //public ca pem file
+	KeyFile       string            // private key pem file
 }

vendor/sparrow-sdk/protocal/protocol.go

@@ -26,3 +26,21 @@ type CloudSend struct {
 	Timestamp   int    `json:"timestamp"`
 	Data        *Data  `json:"data"`
 }
+
+// DevLogin 子设备上线
+type DevLogin struct {
+	Action      string `json:"action"`
+	MsgId       int    `json:"msgId"`
+	DeviceCode  string `json:"deviceCode"`
+	SubDeviceId string `json:"subDeviceId"`
+	Timestamp   int64  `json:"timestamp"`
+}
+
+// DevLogout 子设备下线
+type DevLogout struct {
+	Action      string `json:"action"`
+	MsgId       int    `json:"msgId"`
+	DeviceCode  string `json:"deviceCode"`
+	SubDeviceId string `json:"subDeviceId"`
+	Timestamp   int64  `json:"timestamp"`
+}

vendor/sparrow-sdk/v2/gateway.go

@@ -2,11 +2,13 @@ package v2
 
 import (
 	"context"
+	"crypto/tls"
 	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"fmt"
 	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"github.com/gogf/gf/container/gmap"
 	"github.com/gogf/gf/encoding/gjson"
 	"github.com/gogf/gf/net/ghttp"
 	"log"
@@ -23,11 +25,17 @@ type CmdMessage struct {
 	Cmd    string
 	Params interface{}
 }
+type CmdCallbackFun func(msg protocal.CloudSend) error
 
 // DeviceReportCommandCb 云平台下发的上报指令回调
 type DeviceReportCommandCb func(deviceCode, subId string) error
 
 func NewGateway(config *config.Config) *Gateway {
+	if config.UseTls {
+		if config.CaFile == "" || config.KeyFile == "" {
+			panic("use tls: CaFile and CaKey must be provide")
+		}
+	}
 	c := ghttp.NewClient()
 	c.SetHeader("Content-Type", "application/json")
 	if config.Logger == nil {
@@ -39,6 +47,7 @@ func NewGateway(config *config.Config) *Gateway {
 		httpClient:         c,
 		closeChan:          make(chan struct{}),
 		commandMessageChan: make(chan protocal.CloudSend),
+		cmdList:            gmap.New(true),
 	}
 }
 
@@ -58,6 +67,7 @@ type Gateway struct {
 	closeChan          chan struct{}
 	commandMessageChan chan protocal.CloudSend
 	reportCommandCb    DeviceReportCommandCb
+	cmdList            *gmap.Map
 }
 
 func (a *Gateway) SetReportCommandCallback(cb DeviceReportCommandCb) {
@@ -129,7 +139,13 @@ func (a *Gateway) Authentication() (*schema.DeviceAuthData, error) {
 
 // Connect 接入平台,会阻塞主进程
 func (a *Gateway) Connect() {
-	opts := mqtt.NewClientOptions().AddBroker(fmt.Sprintf("tcp://%s", a.accessAddr))
+	var url string
+	if a.config.UseTls {
+		url = fmt.Sprintf("ssl://%s", a.accessAddr)
+	} else {
+		url = fmt.Sprintf("tcp://%s", a.accessAddr)
+	}
+	opts := mqtt.NewClientOptions().AddBroker(url)
 	clientId := fmt.Sprintf("%x", a.deviceId)
 	opts.SetClientID(clientId)
 	opts.SetPassword(hex.EncodeToString(a.accessToken))
@@ -137,6 +153,13 @@ func (a *Gateway) Connect() {
 	opts.SetOnConnectHandler(func(client mqtt.Client) {
 		a.config.Logger.Trace(context.Background(), "%s", "成功接入平台")
 	})
+	if a.config.UseTls {
+		cert, err := tls.LoadX509KeyPair(a.config.CaFile, a.config.KeyFile)
+		if err != nil {
+			panic(err)
+		}
+		opts.SetTLSConfig(&tls.Config{Certificates: []tls.Certificate{cert}, InsecureSkipVerify: true})
+	}
 	opts.SetConnectionLostHandler(func(client mqtt.Client, err error) {
 		a.config.Logger.Trace(context.Background(), "与平台断开连接[%s]!", err.Error())
 	})
@@ -208,8 +231,15 @@ func (a *Gateway) commandHandler(message mqtt.Message) {
 			if err = a.reportCommandCb(msg.DeviceCode, msg.SubDeviceId); err != nil {
 				panic(err)
 			}
+			return
 		}
 		a.config.Logger.Trace(context.Background(), "gateway receiving command:%+v", msg.Data.Cmd)
+		if a.cmdList.Contains(msg.Data.Cmd) {
+			f := a.cmdList.Get(msg.Data.Cmd)
+			if err = f.(CmdCallbackFun)(msg); err != nil {
+				a.config.Logger.Trace(context.Background(), "执行指令失败:%s", msg.Data.Cmd)
+			}
+		}
 		select {
 		case a.commandMessageChan <- msg:
 		case <-time.After(5 * time.Second):
@@ -221,6 +251,50 @@ func (a *Gateway) commandHandler(message mqtt.Message) {
 }
 
 // RecvCommand recv a command message from channel
+// Deprecated
 func (a *Gateway) RecvCommand() <-chan protocal.CloudSend {
 	return a.commandMessageChan
 }
+
+// RegisterCommand 注册指令回调
+func (a *Gateway) RegisterCommand(cmd string, f CmdCallbackFun) error {
+	if a.cmdList.Contains(cmd) {
+		return errors.New("重复注册")
+	}
+	a.cmdList.Set(cmd, f)
+	return nil
+}
+
+// SubDeviceLogin 子设备上线
+func (a *Gateway) SubDeviceLogin(deviceCode, subDeviceId string) error {
+	data := &protocal.DevLogin{
+		Action:      "devLogin",
+		MsgId:       1,
+		DeviceCode:  deviceCode,
+		SubDeviceId: subDeviceId,
+		Timestamp:   time.Now().Unix(),
+	}
+	payload, err := json.Marshal(data)
+	if err != nil {
+		return err
+	}
+	a.mqttClient.Publish("s", 1, false, payload)
+	return nil
+}
+
+// SubDeviceLogout 子设备下线
+func (a *Gateway) SubDeviceLogout(deviceCode, subDeviceId string) error {
+	data := &protocal.DevLogin{
+		Action:      "devLogout",
+		MsgId:       1,
+		DeviceCode:  deviceCode,
+		SubDeviceId: subDeviceId,
+		Timestamp:   time.Now().Unix(),
+	}
+	payload, err := json.Marshal(data)
+	if err != nil {
+		return err
+	}
+	a.mqttClient.Publish("s", 1, false, payload)
+	return nil
+}