public/PandaX
1
0
Fork 0
mirror of https://gitee.com/XM-GO/PandaX.git synced 2026-04-23 02:48:34 +08:00
Code Issues Packages Projects Releases Wiki Activity

优化

Browse Source
This commit is contained in:
tfl
2024-08-27 12:23:15 +08:00
parent 77ac18c21b
commit 123c304a01
13 changed files with 105 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
pkg/rule_engine/message/message.go
View File

@@ -2,9 +2,11 @@ package message
import (
"encoding/json"
"github.com/google/uuid"
"github.com/sirupsen/logrus"
"time"
"github.com/google/uuid"
"github.com/kakuilan/kgo"
"github.com/sirupsen/logrus"
)
// 消息类型
@@ -14,7 +16,6 @@ const (
RpcRequestFromDevice = "RpcRequestFromDevice"
RpcRequestToDevice = "RpcRequestToDevice"
UpEventMes = "Event"
AlarmMes = "Alarm"
RowMes = "Row"
TelemetryMes = "Telemetry"
AttributesMes = "Attributes"
@@ -117,6 +118,26 @@ func (meta *Metadata) GetValue(key string) any {
}
return (*meta)[key]
}
func (meta *Metadata) GetStringValue(key string) string {
if _, found := (*meta)[key]; !found {
return ""
}
return kgo.KConv.ToStr((*meta)[key])
}
func (meta *Metadata) GetIntValue(key string) int {
if _, found := (*meta)[key]; !found {
return 0
}
return kgo.KConv.ToInt((*meta)[key])
}
func (meta *Metadata) GetFloat64Value(key string) float64 {
if _, found := (*meta)[key]; !found {
return 0
}
return kgo.KConv.ToFloat((*meta)[key])
}
func (meta *Metadata) Has(key string) bool {
_, ok := (*meta)[key]

9
pkg/rule_engine/nodes/action_clear_alarm_node.go
View File

@@ -2,6 +2,7 @@ package nodes
import (
"encoding/json"
"errors"
"pandax/apps/device/services"
"pandax/pkg/global"
"pandax/pkg/rule_engine/message"
@@ -32,8 +33,14 @@ func (n *clearAlarmNode) Handle(msg *message.Message) error {
cleared := n.GetLinkedNode("Cleared")
failure := n.GetLinkedNode("Failure")
var deviceId string
if did, ok := msg.Metadata.GetValue("deviceId").(string); ok {
deviceId = did
} else {
return errors.New("元数据中为获取到设备ID")
}
var err error
alarm, err := services.DeviceAlarmModelDao.FindOneByType(msg.Metadata.GetValue("deviceId").(string), n.AlarmType, "0")
alarm, err := services.DeviceAlarmModelDao.FindOneByType(deviceId, n.AlarmType, "0")
if err == nil {
alarm.State = global.CLEARED
marshal, _ := json.Marshal(msg.Msg)

23
pkg/rule_engine/nodes/action_create_alarm_node.go
View File

@@ -2,9 +2,10 @@ package nodes
import (
"encoding/json"
"github.com/PandaXGO/PandaKit/utils"
"errors"
"pandax/apps/device/entity"
"pandax/apps/device/services"
"pandax/kit/utils"
"pandax/pkg/global"
"pandax/pkg/rule_engine/message"
"time"
@@ -34,8 +35,14 @@ func (n *createAlarmNode) Handle(msg *message.Message) error {
created := n.GetLinkedNode("Created")
updated := n.GetLinkedNode("Updated")
failure := n.GetLinkedNode("Failure")
var deviceId string
if did, ok := msg.Metadata.GetValue("deviceId").(string); ok {
deviceId = did
} else {
return errors.New("元数据中为获取到设备ID")
}
var err error
alarm, err := services.DeviceAlarmModelDao.FindOneByType(msg.Metadata.GetValue("deviceId").(string), n.AlarmType, "0")
alarm, err := services.DeviceAlarmModelDao.FindOneByType(deviceId, n.AlarmType, "0")
if err == nil {
marshal, _ := json.Marshal(msg.Msg)
alarm.Details = string(marshal)
@@ -48,16 +55,16 @@ func (n *createAlarmNode) Handle(msg *message.Message) error {
}
} else {
alarm = &entity.DeviceAlarm{}
alarm.Id = utils.GenerateID("")
alarm.DeviceId = msg.Metadata.GetValue("deviceId").(string)
alarm.ProductId = msg.Metadata.GetValue("productId").(string)
alarm.Name = msg.Metadata.GetValue("deviceName").(string)
alarm.Id = utils.GenerateID()
alarm.DeviceId = msg.Metadata.GetStringValue("deviceId")
alarm.ProductId = msg.Metadata.GetStringValue("productId")
alarm.Name = msg.Metadata.GetStringValue("deviceName")
alarm.Level = n.AlarmSeverity
alarm.State = global.ALARMING
alarm.Type = n.AlarmType
alarm.Time = time.Now()
alarm.OrgId = msg.Metadata.GetValue("orgId").(int64)
alarm.Owner = msg.Metadata.GetValue("owner").(string)
alarm.OrgId = int64(msg.Metadata.GetIntValue("orgId"))
alarm.Owner = msg.Metadata.GetStringValue("owner")
marshal, _ := json.Marshal(msg.Msg)
alarm.Details = string(marshal)
err = services.DeviceAlarmModelDao.Insert(*alarm)

8
pkg/rule_engine/nodes/action_log_node.go
View File

@@ -39,10 +39,10 @@ func (n *logNode) Handle(msg *message.Message) error {
services.RuleChainMsgLogModelDao.Insert(entity.RuleChainMsgLog{
MessageId: msg.Id,
MsgType: msg.MsgType,
DeviceId: msg.Metadata["deviceId"].(string),
OrgId: msg.Metadata["orgId"].(int64),
Owner: msg.Metadata["owner"].(string),
DeviceName: msg.Metadata["deviceName"].(string),
DeviceId: msg.Metadata.GetStringValue("deviceId"),
OrgId: int64(msg.Metadata.GetIntValue("orgId")),
Owner: msg.Metadata.GetStringValue("owner"),
DeviceName: msg.Metadata.GetStringValue("deviceName"),
Ts: msg.Ts,
Content: logMessage,
})

28
pkg/rule_engine/nodes/action_rpc_request_from_device_node.go
View File

@@ -2,11 +2,11 @@ package nodes
import (
"errors"
"github.com/PandaXGO/PandaKit/utils"
"pandax/apps/device/services"
"pandax/iothub/client/mqttclient"
"pandax/iothub/client/tcpclient"
"pandax/iothub/client/udpclient"
"pandax/kit/utils"
devicerpc "pandax/pkg/device_rpc"
"pandax/pkg/global"
"pandax/pkg/rule_engine/message"
@@ -41,10 +41,20 @@ func (n *rpcRequestFromDeviceNode) Handle(msg *message.Message) error {
return errors.New("指令请求格式错误")
}
var deviceId string
if did, ok := msg.Metadata.GetValue("deviceId").(string); ok {
deviceId = did
} else {
return errors.New("元数据中为获取到设备ID")
}
var rpcp = devicerpc.RpcPayload{
Method: msg.Msg.GetValue("method").(string),
Params: msg.Msg.GetValue("params"),
}
if method, ok := msg.Metadata.GetValue("method").(string); ok {
rpcp.Method = method
} else {
return errors.New("指令方法格式错误")
}
var err error
// 指令下发响应
if rpcp.Method == "cmdResp" {
@@ -63,21 +73,19 @@ func (n *rpcRequestFromDeviceNode) Handle(msg *message.Message) error {
}
// 判断设备协议，根据不通协议，发送不通内容
deviceProtocol := global.MQTTProtocol
if msg.Metadata.GetValue("deviceProtocol") != nil && msg.Metadata.GetValue("deviceProtocol").(string) != "" {
if msg.Metadata.GetValue("deviceProtocol") != nil && msg.Metadata.GetStringValue("deviceProtocol") != "" {
deviceProtocol = msg.Metadata.GetValue("deviceProtocol").(string)
}
deviceId := msg.Metadata.GetValue("deviceId").(string)
if deviceProtocol == global.MQTTProtocol || deviceProtocol == global.CoAPProtocol || deviceProtocol == global.LwM2MProtocol {
rpc := &mqttclient.RpcRequest{}
RequestId := n.RequestId
if RequestId == "" {
if msg.Metadata.GetValue("requestId") == nil {
rpc.RequestId = utils.GenerateID("")
if n.RequestId == "" {
if msg.Metadata.GetStringValue("requestId") == "" {
rpc.RequestId = utils.GenerateID()
} else {
rpc.RequestId = msg.Metadata.GetValue("requestId").(string)
rpc.RequestId = msg.Metadata.GetStringValue("requestId")
}
} else {
rpc.RequestId = RequestId
rpc.RequestId = n.RequestId
}
err = rpc.Pub(deviceId, result)
}

25
pkg/rule_engine/nodes/action_rpc_request_to_device_node.go
View File

@@ -3,12 +3,12 @@ package nodes
import (
"encoding/json"
"errors"
"github.com/PandaXGO/PandaKit/utils"
"pandax/apps/device/entity"
"pandax/apps/device/services"
"pandax/iothub/client/mqttclient"
"pandax/iothub/client/tcpclient"
"pandax/iothub/client/udpclient"
"pandax/kit/utils"
"pandax/pkg/global"
"pandax/pkg/global/model"
"pandax/pkg/rule_engine/message"
@@ -41,21 +41,32 @@ func (n *rpcRequestToDeviceNode) Handle(msg *message.Message) error {
if msg.Msg.GetValue("method") == nil || msg.Msg.GetValue("params") == nil {
return errors.New("指令下发格式错误")
}
deviceId := msg.Metadata.GetValue("deviceId").(string)
var deviceId string
if did, ok := msg.Metadata.GetValue("deviceId").(string); ok {
deviceId = did
} else {
return errors.New("元数据中为获取到设备ID")
}
// 创建请求格式
var datas = model.RpcPayload{
Method: msg.Msg.GetValue("method").(string),
Params: msg.Msg.GetValue("params"),
}
if method, ok := msg.Metadata.GetValue("method").(string); ok {
datas.Method = method
} else {
return errors.New("指令方法格式错误")
}
payload, _ := json.Marshal(datas)
// 构建指令记录
var data entity.DeviceCmdLog
data.Id = utils.GenerateID("")
data.Id = utils.GenerateID()
data.DeviceId = deviceId
data.CmdName = datas.Method
data.CmdContent = kgo.KConv.ToStr(datas.Params)
data.Mode = msg.Metadata.GetValue("mode").(string)
if mode, ok := msg.Metadata.GetValue("mode").(string); ok {
data.Mode = mode
}
data.State = "2"
data.RequestTime = time.Now().Format("2006-01-02 15:04:05")
@@ -65,8 +76,8 @@ func (n *rpcRequestToDeviceNode) Handle(msg *message.Message) error {
}
// 判断设备协议，根据不通协议，发送不通内容
deviceProtocol := global.MQTTProtocol
if msg.Metadata.GetValue("deviceProtocol") != nil && msg.Metadata.GetValue("deviceProtocol").(string) != "" {
deviceProtocol = msg.Metadata.GetValue("deviceProtocol").(string)
if msg.Metadata.GetValue("deviceProtocol") != nil && msg.Metadata.GetStringValue("deviceProtocol") != "" {
deviceProtocol = msg.Metadata.GetStringValue("deviceProtocol")
}
var err error
if deviceProtocol == global.MQTTProtocol || deviceProtocol == global.CoAPProtocol || deviceProtocol == global.LwM2MProtocol {

8
pkg/rule_engine/nodes/action_save_attributes_node.go
View File

@@ -1,6 +1,7 @@
package nodes
import (
"errors"
"pandax/pkg/global"
"pandax/pkg/rule_engine/message"
)
@@ -33,7 +34,12 @@ func (n *saveAttributesNode) Handle(msg *message.Message) error {
}
}*/
//deviceId := msg.GetMetadata().GetValues()["deviceId"].(string)
deviceName := msg.Metadata["deviceName"].(string)
var deviceName string
if dn, ok := msg.Metadata.GetValue("deviceName").(string); ok {
deviceName = dn
} else {
return errors.New("元数据中为获取到设备ID")
}
err := global.TdDb.InsertDevice(deviceName+"_attributes", msg.Msg)
if err != nil {
n.Debug(msg, message.DEBUGOUT, err.Error())

8
pkg/rule_engine/nodes/action_save_timeseries_node.go
View File

@@ -1,6 +1,7 @@
package nodes
import (
"errors"
"pandax/pkg/global"
"pandax/pkg/rule_engine/message"
)
@@ -33,7 +34,12 @@ func (n *saveTimeSeriesNode) Handle(msg *message.Message) error {
return nil
}
}*/
deviceName := msg.Metadata["deviceName"].(string)
var deviceName string
if dn, ok := msg.Metadata.GetValue("deviceName").(string); ok {
deviceName = dn
} else {
return errors.New("元数据中为获取到设备ID")
}
err := global.TdDb.InsertDevice(deviceName+"_telemetry", msg.Msg)
if err != nil {
n.Debug(msg, message.DEBUGOUT, err.Error())

2
pkg/rule_engine/nodes/external_ding_node.go
View File

@@ -6,8 +6,8 @@ import (
"encoding/base64"
"encoding/json"
"fmt"
"github.com/PandaXGO/PandaKit/httpclient"
"net/url"
"pandax/kit/httpclient"
"pandax/pkg/rule_engine/message"
"time"
)

2
pkg/rule_engine/nodes/external_restapi_node.go
View File

@@ -3,7 +3,7 @@ package nodes
import (
"encoding/json"
"errors"
"github.com/PandaXGO/PandaKit/httpclient"
"pandax/kit/httpclient"
"pandax/pkg/rule_engine/message"
)

2
pkg/rule_engine/nodes/external_wechat_node.go
View File

@@ -2,7 +2,7 @@ package nodes
import (
"encoding/json"
"github.com/PandaXGO/PandaKit/httpclient"
"pandax/kit/httpclient"
"pandax/pkg/rule_engine/message"
)

3
pkg/rule_engine/nodes/filter_device_type_switch_node.go
View File

@@ -29,8 +29,7 @@ func (n *deviceTypeSwitchNode) Handle(msg *message.Message) error {
deviceLabelNode := n.GetLinkedNode(message.DEVICE)
gatewayLabelNode := n.GetLinkedNode(message.GATEWAY)
deviceType := msg.Metadata.GetValue("deviceType").(string)
deviceType := msg.Metadata.GetStringValue("deviceType")
if deviceType == message.DEVICE {
if deviceLabelNode != nil {
n.Debug(msg, message.DEBUGOUT, "")

2
pkg/rule_engine/nodes/filter_message_type_switch_node.go
View File

@@ -18,10 +18,8 @@ func (f messageTypeSwitchNodeFactory) Labels() []string {
message.TelemetryMes,
message.RpcRequestFromDevice,
message.RpcRequestToDevice,
message.AlarmMes,
message.UpEventMes,
message.ConnectMes,
message.ConnectMes,
message.DisConnectMes,
}
}