sparrow/services/scene-service/internal/service/manager/device_status.go

package manager

import (
	"encoding/json"
	"errors"
	"fmt"
	"github.com/gogf/gf/container/gmap"
	"github.com/gogf/gf/encoding/gjson"
	"sparrow/pkg/rpcs"
	"sparrow/pkg/server"
	"sparrow/pkg/utils"
	"time"
)

type DeviceSceneConfig struct {
	SceneId      string             `json:"scene_id"`
	DecisionExpr string             `json:"decision_expr"` // 条件表达式 and or
	Conditions   []*DeviceCondition `json:"conditions"`    // 条件
	Actions      []*Action          `json:"actions"`       // 执行动作
	Interval     int                `json:"interval"`      // 检查间隔(秒)
	ticker       *time.Ticker       `json:"-"`             // 定时器
	stopChan     chan struct{}      `json:"-"`             // 停止信号通道
}

// DeviceCondition 设备场景配置
type DeviceCondition struct {
	ConditionId string `json:"condition_id"`  // 条件id
	Key         string `json:"key"`           // redis key
	DeviceType  string `json:"device_type"`   // 设备类型
	DeviceId    string `json:"device_id"`     // 设备id
	SubDeviceId string `json:"sub_device_id"` // 子设备id
	FieldName   string `json:"field_name"`    // 字段名称
	FieldType   int    `json:"field_type"`    // 字段类型 1字符串 2数值
	Field       string `json:"field"`         // 字段名
	TargetValue string `json:"target_value"`  // 值
	Operator    int    `json:"operator"`      // 比较类型  数值比较 1 >  2 >= 3 = 4 <= 5 < 6 !=
}

type DeviceSceneService struct {
	tasks *gmap.HashMap
}

func NewDeviceSceneService() *DeviceSceneService {
	return &DeviceSceneService{
		tasks: gmap.New(true),
	}
}

func (d *DeviceSceneService) Add(config string) error {
	var c DeviceSceneConfig
	err := json.Unmarshal([]byte(config), &c)
	if err != nil {
	}
	if len(c.Conditions) == 0 {
		return errors.New("设备状态监控任务配置错误：判断条件不能为空")
	}
	// 初始化Ticker和停止通道
	c.ticker = time.NewTicker(600 * time.Second)
	c.stopChan = make(chan struct{})

	// 启动监控协程
	go d.monitorTask(c)
	fmt.Printf("保存设备状态监控任务成功：%s\n", gjson.New(c).MustToJsonString())
	d.tasks.Set(c.SceneId, c)
	return nil
}

func (d *DeviceSceneService) Update(config string) error {
	var c DeviceSceneConfig
	err := json.Unmarshal([]byte(config), &c)
	if err != nil {
		server.Log.Errorf("config to timerConfig error :%s", err.Error())
	}
	_ = d.Stop(c.SceneId)

	// 初始化Ticker和停止通道
	c.ticker = time.NewTicker(600 * time.Second)
	c.stopChan = make(chan struct{})

	// 启动监控协程
	go d.monitorTask(c)
	d.tasks.Set(c.SceneId, c)
	server.Log.Debugf("UpdateWeatherScene :%s", config)
	return nil
}

func (d *DeviceSceneService) Remove(id string) error {
	scene := d.tasks.Get(id)
	if scene == nil {
		return nil
	}
	c := scene.(DeviceSceneConfig)
	c.stopChan <- struct{}{}
	d.tasks.Remove(c.SceneId)
	return nil
}

// Start 停止任务
func (d *DeviceSceneService) Start(id string) error {
	if !d.tasks.Contains(id) {
		return errors.New("任务不存在")
	}
	task := d.tasks.Get(id)
	c := task.(DeviceSceneConfig)
	go d.monitorTask(c)
	return nil
}

// Stop 停止任务
func (d *DeviceSceneService) Stop(id string) error {
	if !d.tasks.Contains(id) {
		return errors.New("任务不存在")
	}
	task := d.tasks.Get(id)
	c := task.(DeviceSceneConfig)
	c.stopChan <- struct{}{}
	return nil
}

// monitorTask 监控任务：使用select监听Ticker和停止信号
func (d *DeviceSceneService) monitorTask(config DeviceSceneConfig) {
	for {
		select {
		case <-config.ticker.C: // 定时触发
			result, err := d.checkDeviceCondition(config)
			if err != nil {
				server.Log.Errorf("compare weather condition error :%s", err.Error())
			}
			if result.Result {
				taskExecutor := NewTaskExecutor(config.Actions)
				if err = taskExecutor.Do(config.SceneId); err != nil {
					server.Log.Errorf("device_status do taskid :%s error:%s", config.SceneId, err.Error())
				}
				_ = taskExecutor.saveHis(config.SceneId, result.ConditionId)
			}

		case <-config.stopChan: // 收到停止信号
			config.ticker.Stop()
			return
		}
	}
}

func (d *DeviceSceneService) checkDeviceCondition(config DeviceSceneConfig) (CheckResult, error) {
	var results []bool
	var checkResult CheckResult
	var err error
	for _, v := range config.Conditions {

		var args rpcs.ArgsGetStatus
		args.Key = v.Key
		var reply rpcs.ReplyStatus
		err := server.RPCCallByName(nil, rpcs.DeviceManagerName, "DeviceManager.GetDeviceStatusByKey", args, &reply)
		if err != nil {
			server.Log.Errorf("设备状态数据获取失败：%v", err)
			return checkResult, err
		}
		j := gjson.New(reply.Status)
		// 判断是否满足条件并填入到result
		result := utils.CheckValue(v.TargetValue, j.Get(v.Field), v.FieldType, v.Operator)
		if result {
			checkResult.ConditionId = append(checkResult.ConditionId, v.ConditionId)
		}
		results = append(results, result)

	}
	switch config.DecisionExpr {
	case "and":
		for _, v := range results {
			if !v {
				checkResult.Result = false
			}
		}
		checkResult.Result = true
	case "or":
		for _, v := range results {
			if v {
				checkResult.Result = true
			}
		}
	default:
		err = errors.New("无效的判断逻辑")
	}

	return checkResult, err
}