车间设备接口

@@ -2,15 +2,18 @@ package constant
 
 import "os"
 
+var MQTT_TOPIC = "/MQTT"
 //设备商提供的测试地址
-var MQTT_HOST = "175.24.122.87"
+//var MQTT_HOST = "175.24.122.87"
+//var MQTT_PORT = "1883"
+//var MQTT_USER = "user111"
+//var MQTT_PASSWORD = "user111"
 //内网测试地址
-//var MQTT_HOST = "192.168.100.222"
+var MQTT_HOST = "192.168.100.222"
 var MQTT_PORT = "1883"
+var MQTT_USER = "admin"
+var MQTT_PASSWORD = "123456"
 
-var MQTT_USER = ""
-
-var MQTT_PASSWORD = ""
 
 func init(){
 	if os.Getenv("MQTT_HOST") != "" {

@@ -14,6 +14,8 @@ var (
 	REDIS_ADDRESS = ""
 	// redis 考勤机打卡消息队列
 	REDIS_ZKTECO_KEY = "allied-creation-zkteco"
+	// redis 车间数据消息队列
+	REDIS_WORKSHOP_KEY = "allied-creation-workshop"
 )
 
 func init() {

+package domain
+
+import "time"
+
+type DeviceWorkShop struct {
+	WorkShop     string    `json:"WorkShop"`     // 车间名
+	DeviceSn     string    `json:"DeviceSn"`     // 设备名称
+	CurrTime     time.Time `json:"CurrTime"`     // 当前时间
+	StartupState int64     `json:"StartupState"` // 启动状态：1：启动，0：停止
+	ComStatus    int64     `json:"ComStatus"`    // 通讯状态：1：通讯正常，0：设备未上电或与采集端通讯故障
+	InterSpeed   int64     `json:"InterSpeed"`   // 内包材速度：内包材运行速率
+	ExterSpeed   int64     `json:"ExterSpeed"`   // 外包材速度：内包材运行速率
+	KnifeSpeed   int64     `json:"KnifeSpeed"`   // 切刀速度：切刀运行速率
+	TransSpeed   int64     `json:"TransSpeed"`   // 输送速度：输送带运行速率
+	FrontTemp    float64   `json:"FontTemp"`     // 炸机前段温度：炸机前段当前温度
+	BackTemp     float64   `json:"BackTemp"`     // 炸机后段温度：炸机后段当前温度
+	TankTemp     float64   `json:"TankTemp"`     // 储油罐温度 ：储油罐当前温度
+	TubeTemp     float64   `json:"TubeTemp"`     // 管路温度：管路当前温度
+	Temp1        float64   `json:"Temp1"`        // 温度1：温度1当前温度
+	Temp2        float64   `json:"Temp2"`        // 温度2：温度2当前温度
+	Temp3        float64   `json:"Temp3"`        // 温度3：温度3当前温度
+	Year         string    `json:"Year"`         // 年
+	Month        string    `json:"Month"`        // 月
+	Day          string    `json:"Day"`          // 日
+	ProductType  string    `json:"ProductType"`  // 产品类型：当前产品种类
+	CurrTemp     float64   `json:"CurrTemp"`     // 当前温度：当前温度
+}

@@ -25,6 +25,9 @@ func (subscribeClient *SubscribeClient) options() *pahomqtt.ClientOptions {
 	opts.SetPassword(constant.MQTT_PASSWORD)
 	opts.SetKeepAlive(2 * time.Second)
 	opts.SetPingTimeout(1 * time.Second)
+	opts.CleanSession = false
+	opts.SetClientID("test")
+	//opts.Order = true
 	return opts
 }
 

@@ -37,11 +37,12 @@ func (controller *DeviceZKTecoController) PostCdata() {
 
 func (controller *DeviceZKTecoController) GetCdata() {
 	//sn := controller.Ctx.Input.Query("SN")
-
+	controller.Ctx.WriteString("OK")
 }
 
 func (controller *DeviceZKTecoController) GetRequest() {
-
+	//controller.Ctx.WriteString("C:11:DATA\tQUERY\tUSERINFO\tPIN=10086")
+	controller.Ctx.WriteString("OK")
 }
 
 func (controller *DeviceZKTecoController) Ping() {

 package mqtt
 
 import (
-	"fmt"
+	"encoding/json"
 	pahomqtt "github.com/eclipse/paho.mqtt.golang"
+	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/constant"
+	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/domain"
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/infrastructure/mqtt"
-	"strconv"
+	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/infrastructure/redis"
+	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/log"
 	"time"
 )
 
-func Start(){
-	mqtt.StartSubscribe("test", func(client pahomqtt.Client, message pahomqtt.Message) {
-		fmt.Println(time.Now(),"Topic:"+message.Topic()+" MessageId:"+ strconv.Itoa(int(message.MessageID()))+" Message:"+ string(message.Payload()))
+func Start() {
+	mqtt.StartSubscribe(constant.MQTT_TOPIC, func(client pahomqtt.Client, message pahomqtt.Message) {
+		payload := make(map[string]interface{})
+		err := json.Unmarshal(message.Payload(), &payload)
+		if err != nil {
+			log.Logger.Error("车间数据json解析失败：" + err.Error())
+			return
+		}
+		if workShop, ok := payload["WorkShop"]; ok {
+			for key, item := range payload {
+				if key == "WorkShop" {
+					continue
+				}
+				deviceWorkShop := &domain.DeviceWorkShop{}
+				mBytes, err := json.Marshal(item)
+				if err != nil {
+					continue
+				}
+				err = json.Unmarshal(mBytes, deviceWorkShop)
+				if err != nil {
+					continue
+				}
+				// 获取当前时间
+				deviceWorkShop.CurrTime = time.Now()
+				// 车间名称
+				deviceWorkShop.WorkShop = workShop.(string)
+				// 设备名称
+				deviceWorkShop.DeviceSn = key
+				workShopBytes, err := json.Marshal(deviceWorkShop)
+				if err != nil {
+					continue
+				}
+				err = redis.GetRedis().LPush(constant.REDIS_WORKSHOP_KEY, string(workShopBytes)).Err()
+				if err != nil {
+					log.Logger.Error("车间设备数据加入redis失败：" + err.Error())
+				}
+			}
+		}
+		log.Logger.Info("MQTT", map[string]interface{}{
+			"Topic":     message.Topic(),
+			"MessageId": message.MessageID(),
+			"Message":   payload,
+		})
 	})
-}
\ No newline at end of file
+}