confrontation-training/initialize/emq.go

package initialize

import (
	gateway2 "confrontation-training/api/gateway"
	"confrontation-training/constant"
	"confrontation-training/global"
	"confrontation-training/models/emq"
	"confrontation-training/models/gateway"
	"encoding/json"
	"fmt"
	mqtt "github.com/eclipse/paho.mqtt.golang"
	"github.com/gorilla/websocket"
	"log"
	"strconv"
	"strings"
	"time"
)

func CreateEmqClient() {

	config := global.Config
	global.EmqConfig.ClientId = config.EmqConfig.ClientId
	global.EmqConfig.Qos = config.EmqConfig.Qos
	global.EmqConfig.Topic = config.EmqConfig.Topic
	global.EmqConfig.UserName = config.EmqConfig.UserName
	global.EmqConfig.Password = config.EmqConfig.Password
	global.EmqConfig.Protocol = config.EmqConfig.Protocol
	global.EmqConfig.Port = config.EmqConfig.Port
	global.EmqConfig.Broker = config.EmqConfig.Broker
	global.EmqConfig.Filter = config.EmqConfig.Filter
	global.EmqConfig.GatewayMac = config.EmqConfig.GatewayMac
	emqConfig := global.EmqConfig
	connectAddress := fmt.Sprintf("%s://%s:%d", emqConfig.Protocol, emqConfig.Broker, emqConfig.Port)
	fmt.Println("connect address:", connectAddress)
	opts := mqtt.NewClientOptions()
	opts.AddBroker(connectAddress)
	opts.SetUsername(emqConfig.UserName)
	opts.SetPassword(emqConfig.Password)
	opts.SetClientID(emqConfig.ClientId)
	opts.SetKeepAlive(time.Second * 60)
	global.EmqClient = mqtt.NewClient(opts)
	client := global.EmqClient
	token := client.Connect()
	if token.WaitTimeout(time.Second*3) && token.Error() != nil {
		log.Fatal(token.Error())
	}
	SubScribe(client)
	//SendUUID(client, "/EE3870DA24C4/connect_packet/connect1_subscribe")
	//Publish(client, "/EE3870DA24C4/connect_packet/connect1_subscribe", constant.CmdStartScan)
	//SendUUID(client, "/EE3870DA24C4/connect_packet/connect1_subscribe")
	//读取UUID
	//ReadUUID(client, "/EE3870DA24C4/connect_packet/connect1_subscribe")
	//连接设备
	//ConnectDevice(client, "/EE3870DA24C4/connect_packet/connect1_subscribe", "DC0D30033AF0", "0", "0", "0")

	//fmt.Println("link device DC0D30033AF0 over")
	//ConnectDevice(client, "/EE3870DA24C4/connect_packet/connect1_subscribe", "DC0D300335FC", "0", "0", "0")
}

func Publish(client mqtt.Client, topic string, message string) {

	if token := client.Publish(topic, byte(global.EmqConfig.Qos), false, message); token.Wait() && token.Error() != nil {
		fmt.Printf("publish failed, topic: %s, payload: %s\n", topic, message)
	} else {
		fmt.Printf("publish success, topic: %s, payload: %s\n", topic, message)
	}
	//time.Sleep(time.Second * 1)
	time.Sleep(time.Millisecond * 100)
}
func SendUUID(client mqtt.Client, topic string) {

	if token := client.Publish(topic, byte(global.EmqConfig.Qos), false, constant.CmdSetUUID); token.Wait() && token.Error() != nil {
		fmt.Printf("publish failed, topic: %s, payload: %s\n", topic, constant.CmdSetUUID)
	} else {
		fmt.Printf("publish success, topic: %s, payload: %s\n", topic, constant.CmdSetUUID)
	}
	//time.Sleep(time.Second * 1)
	time.Sleep(time.Millisecond * 100)
}
func ConnectDevice(client mqtt.Client, topic string, mac string, index string, ai string, at string) {
	message := "[{\"cmd\":\"AT+CNN=\",\"m\":\"" + mac + "\",\"i\":\"" + index + "\",\"ai\":\"" + ai + "\",\"at\":\"" + at + "\",\"l\":\"1\",\"x\":\"251\",\"relink\":\"1\",\"timeout\":\"12000\"}]"
	if token := client.Publish(topic, byte(global.EmqConfig.Qos), false, message); token.Wait() && token.Error() != nil {
		fmt.Printf("publish failed, topic: %s, payload: %s\n", topic, message)
	} else {
		fmt.Printf("publish success, topic: %s, payload: %s\n", topic, message)
	}
	//time.Sleep(time.Second * 1)
	time.Sleep(time.Millisecond * 100)
}
func SubScribe(client mqtt.Client) {
	emqConfig := global.EmqConfig
	ws, _, err := websocket.DefaultDialer.Dial(global.Config.Websocket.WSUrl, nil)
	if err != nil {
		fmt.Println("socket通道初始化失败")
		panic(err)
		return
	}
	deviceService := GetDeviceService()
	d, i := deviceService.FindDeviceByType("")
	global.DeviceMap = make(map[string]gateway.DeviceInfo)
	if i > 0 {
		for j := 0; j < len(d); j++ {
			global.DeviceMap[d[j].Mac] = d[j]
		}
	}
	for _, topic := range emqConfig.Topic {
		topic = "/" + global.EmqConfig.GatewayMac + topic
		client.Subscribe(topic, byte(emqConfig.Qos), func(client mqtt.Client, message mqtt.Message) {
			//fmt.Printf("`%s` Received `%s` from `%s` topic\n", common.NowTime("2006-01-02 15:04:05"), message.Payload(), message.Topic())
			messageMap := make(map[string]string)
			var payloads []emq.Payload
			err := json.Unmarshal(message.Payload(), &payloads)
			if err != nil {
				_ = fmt.Errorf("%s", err.Error())
				return
			}
			for _, payload := range payloads {
				messageMap["state"] = payload.State
				content := ""
				sendFlag := true
				switch payload.Cmd {
				case constant.PayloadScan:
					{
						//关闭扫描
						if payload.S == "0" {
							messageMap["msgType"] = constant.MessageTypeCloseScan
							content = "关闭扫描"
							//开启扫描
						} else if payload.S == "1" {
							messageMap["msgType"] = constant.MessageTypeStartScan
							content = "开启扫描"
						} else if payload.S == "" { //接收到蓝牙扫描数据
							if strings.HasPrefix(payload.N, "MIND") {
								payload.T = "0"
							} else if strings.HasPrefix(payload.N, "BW-ECG") {
								payload.T = "1"
							}
							messageMap["msgType"] = constant.MessageTypeDeviceScanned
							marshal, _ := json.Marshal(payload)
							content = string(marshal)
						}
					}

				case constant.PayloadFLT:
					messageMap["msgType"] = constant.MessageTypeFilter
					content = "设置过滤"
				case constant.PayloadSUUID:
					messageMap["msgType"] = constant.MessageTypeSUUID
					content = "设置UUID"
				case constant.PayloadConnect:
					messageMap["msgType"] = constant.MessageTypeConnect
					content = "设备连接"
				case constant.PayloadDisConnect:
					messageMap["msgType"] = constant.MessageTypeDisConnect
					//content = "断开设备连接"
					marshal, _ := json.Marshal(payload)
					content = string(marshal)
				case constant.PayloadQuenue:
					messageMap["msgType"] = constant.MessageTypeConnectList
					if payload.M == "" {
						sendFlag = false
					} else {
						realMac := transMac(payload.M)
						deviceInfo := global.DeviceMap[realMac]
						if deviceInfo.ID == 0 {
							continue
						} else {
							payload.T = deviceInfo.Type
							payload.N = deviceInfo.Name
							marshal, _ := json.Marshal(payload)
							content = string(marshal)
						}

					}
				case constant.PayloadCNB:
					connectedNum, _ := strconv.Atoi(payload.N)
					for i := 0; i < connectedNum; i++ {
						topic := "/" + global.Config.EmqConfig.GatewayMac + constant.TopicConnectSub
						gateway2.ConnectedListEmq(global.EmqClient, topic, strconv.Itoa(i))
					}
				case constant.PayloadNotify: //接收到蓝牙通知数据
					//1.判断是脑电数据还是心电数据
					//1.1脑电数据直接转发
					//1.2心电数据做处理
					realMac := transMac(payload.M)
					deviceInfo := global.DeviceMap[realMac]
					var receiveData gateway.DeviceDataReceived
					receiveData.Mac = payload.M
					receiveData.Value = payload.D
					if deviceInfo.Type == "0" { //脑电
						messageMap["msgType"] = constant.MessageTypeEEGData
						marshal, _ := json.Marshal(receiveData)
						content = string(marshal)
					} else if deviceInfo.Type == "1" { //心电
						messageMap["msgType"] = constant.MessageTypeECGData
						flag := strings.HasPrefix(receiveData.Value, "E840") || strings.HasPrefix(receiveData.Value, "E841") || strings.HasPrefix(receiveData.Value, "E823") || strings.HasPrefix(receiveData.Value, "E820") || strings.HasPrefix(receiveData.Value, "E81F") || strings.HasPrefix(receiveData.Value, "E813") || strings.HasPrefix(receiveData.Value, "E810") || strings.HasPrefix(receiveData.Value, "E822") || strings.HasPrefix(receiveData.Value, "E826") || strings.HasPrefix(receiveData.Value, "E8FF00000000")
						if !flag {
							//var ecgData []int
							ecgData := [12]int{}
							data := []byte(receiveData.Value[4:])
							if len(data) == 36 {
								for i := 0; i < 36; i += 3 {
									ecgData[i/3] = int(16<<data[i]&0xFF | 8<<data[i+1]&0xFF | data[i+2]&0xFF)
								}
							} else {

								for len(data) < 36 {
									for i := len(data); i < 36; i++ {
										data = append(data, data[i])
									}
								}
								for i := 0; i < 36; i += 3 {
									ecgData[i/3] = int(16<<data[i]&0xFF | 8<<data[i+1]&0xFF | data[i+2]&0xFF)
								}
							}
							messageMap["msgType"] = constant.MessageTypeECGData
							receiveData.Value = "["
							for _, datum := range ecgData {
								receiveData.Value = receiveData.Value + strconv.Itoa(datum) + ","
							}
							receiveData.Value = receiveData.Value + "]"
							marshal, _ := json.Marshal(receiveData)
							content = string(marshal)
						}
					}
				}
				if messageMap["msgType"] != constant.MessageTypeECGData && messageMap["msgType"] != constant.MessageTypeEEGData && messageMap["msgType"] != constant.MessageTypeDeviceScanned && messageMap["msgType"] != constant.MessageTypeConnectList && messageMap["msgType"] != constant.MessageTypeDisConnect {
					if payload.State == constant.Success {
						content = content + "成功"
					} else {
						content = content + "失败"
					}
				}

				messageMap["content"] = content
				messageMap["Sender"] = "server"
				messageMap["Recipient"] = "client"
				bytes, err := json.Marshal(messageMap)
				if err != nil {
					return
				}
				if sendFlag && content != "成功" && content != "失败" {
					err = ws.WriteMessage(websocket.TextMessage, bytes)
					if err != nil {
						fmt.Println("消息发送异常：" + err.Error())
					}
				}
				time.Sleep(time.Millisecond * 5)
			}
		})
	}
}

func transMac(mac string) string {
	result := ""
	for i := 0; i < len(mac); i = i + 2 {
		result = result + mac[i:i+2] + ":"
	}
	return result[0 : len(result)-1]
}