fix: envItem增加adjust方案

envitem/func.go

@@ -22,6 +22,7 @@ func SetOptions(options Options) {
 	}
 
 	fetchMultiItem = options.FetchMultiItem
+	adjustValue = options.AdjustValue
 }
 
 func (m MultiEnvItem) GetProjectId() int64 {
@@ -219,6 +220,13 @@ func (e *EnvItem) getCurrentData() (*ItemValueResp, error) {
 			return nil, errors.New("not found envitem's value")
 		}
 
+		if adjustValue && cache != nil && len(res.Data) > 0 {
+			if rv, err := strconv.ParseFloat(res.Data[0].Val, 64); err == nil {
+				adjust, _ := e.GetAdjustFloat64Val()
+				res.Data[0].Val = fmt.Sprintf("%f", rv+adjust)
+			}
+		}
+
 		return res.Data[0], nil
 	}
 
@@ -359,6 +367,65 @@ func (e *EnvItem) GetItemPrevHtime() *time.Time {
 	return nil
 }
 
+func (e *EnvItem) GetAdjustInt64Val() (int64, error) {
+	return getAdjustInt64Val(e.ProjectId, e.Item)
+}
+
+func (e *EnvItem) GetAdjustFloat64Val() (float64, error) {
+	return getAdjustFloat64Val(e.ProjectId, e.Item)
+}
+
+func (e *EnvItem) GetAdjustStringVal() (string, error) {
+	return getAdjustStringVal(e.ProjectId, e.Item)
+}
+
+func (e *EnvItem) SetAdjust(value string, expire time.Duration) error {
+	if adjustValue == false {
+		return nil
+	}
+
+	if cache == nil {
+		return errors.New("not cache")
+	}
+
+	key := fmt.Sprintf(CACHE_ADJUST_VALUE_KEY, e.ProjectId, e.Item)
+	scmd := cache.Set(context.Background(), key, value, expire)
+	return scmd.Err()
+}
+
+func (e *EnvItem) IncreAdjust(expire time.Duration) (int64, error) {
+	if adjustValue == false {
+		return 0, nil
+	}
+
+	if cache == nil {
+		return 0, errors.New("not cache")
+	}
+
+	key := fmt.Sprintf(CACHE_ADJUST_VALUE_KEY, e.ProjectId, e.Item)
+	intCmd := cache.Incr(context.Background(), key)
+	if intCmd.Err() != nil {
+		return 0, intCmd.Err()
+	}
+
+	cache.Expire(context.Background(), key, expire)
+
+	return intCmd.Val(), nil
+}
+
+func (e *EnvItem) ClearAdjust() {
+	if adjustValue == false {
+		return
+	}
+
+	if cache == nil {
+		return
+	}
+
+	key := fmt.Sprintf(CACHE_ADJUST_VALUE_KEY, e.ProjectId, e.Item)
+	cache.Del(context.Background(), key)
+}
+
 func (e *EnvItem) clearValue() {
 	e.Value = nil
 	e.Htime = ""
@@ -373,3 +440,54 @@ func (d *MultiEnvItem) Scan(input interface{}) error {
 func (d MultiEnvItem) Value() (driver.Value, error) {
 	return json.Marshal(d)
 }
+
+func getAdjustInt64Val(projectId int64, item string) (int64, error) {
+	if adjustValue == false {
+		return 0, nil
+	}
+
+	if cache == nil {
+		return 0, errors.New("not cache")
+	}
+
+	key := fmt.Sprintf(CACHE_ADJUST_VALUE_KEY, projectId, item)
+	scmd := cache.Get(context.Background(), key)
+	if scmd != nil && scmd.Err() != nil {
+		return 0, scmd.Err()
+	}
+	return scmd.Int64()
+}
+
+func getAdjustFloat64Val(projectId int64, item string) (float64, error) {
+	if adjustValue == false {
+		return 0, nil
+	}
+
+	if cache == nil {
+		return 0, errors.New("not cache")
+	}
+
+	key := fmt.Sprintf(CACHE_ADJUST_VALUE_KEY, projectId, item)
+	scmd := cache.Get(context.Background(), key)
+	if scmd != nil && scmd.Err() != nil {
+		return 0, scmd.Err()
+	}
+	return scmd.Float64()
+}
+
+func getAdjustStringVal(projectId int64, item string) (string, error) {
+	if adjustValue == false {
+		return "", nil
+	}
+
+	if cache == nil {
+		return "", errors.New("not cache")
+	}
+
+	key := fmt.Sprintf(CACHE_ADJUST_VALUE_KEY, projectId, item)
+	scmd := cache.Get(context.Background(), key)
+	if scmd != nil && scmd.Err() != nil {
+		return "", scmd.Err()
+	}
+	return scmd.String(), nil
+}

envitem/func_test.go

@@ -2,6 +2,7 @@ package envitem
 
 import (
 	"encoding/json"
+	"github.com/go-redis/redis/v8"
 	"testing"
 	"time"
 )
@@ -20,7 +21,8 @@ func TestEnvItem_GetCurrentData(t *testing.T) {
 }
 
 func TestMultiEnvItem_FillCurrentValue(t *testing.T) {
-	SetOptions(Options{GtServerIp: "47.96.12.136:8788", FetchMultiItem: false})
+	cache := redis.NewClient(&redis.Options{Addr: "47.96.12.136:6379", Password: "", MaxRetries: 5})
+	SetOptions(Options{GtServerIp: "47.96.12.136:8788", FetchMultiItem: false, Cache: cache, AdjustValue: true})
 	m := make(MultiEnvItem, 2)
 	m["C.M.LT_CIP@out"] = &EnvItem{
 		ProjectId: 92,
@@ -31,16 +33,24 @@ func TestMultiEnvItem_FillCurrentValue(t *testing.T) {
 		Item:      "C.M.RO1_DB@time_CS_display",
 	}
 
+	m["C.M.LT_QSC@out"] = &EnvItem{
+		ProjectId: 92,
+		Item:      "C.M.LT_QSC@out",
+	}
+
 	err := m.FillCurrentValue()
 	t.Log(err)
 
 	for s, item := range m {
 		t.Logf("item: %s value: %s htime: %s", s, item.Value, item.Htime)
 	}
+
+	m["C.M.LT_QSC@out"].ClearAdjust()
 }
 
 func TestMultiEnvItem_FillCurrentValue2(t *testing.T) {
-	SetOptions(Options{GtServerIp: "47.96.12.136:8788", FetchMultiItem: false})
+	cache := redis.NewClient(&redis.Options{Addr: "47.96.12.136:6379", Password: "", MaxRetries: 5})
+	SetOptions(Options{GtServerIp: "47.96.12.136:8788", FetchMultiItem: false, Cache: cache})
 	js := `{"tmp":{"project_id":92,"device_code":"UF-4001A","plc_device_id":"0","item":"C.M.UF1_DB@press_PV","default_val":0},"feed_flow":{"project_id":92,"device_code":"UF-4001A","plc_device_id":"0","item":"C.M.UF1_FT_JS@out","default_val":0},"feed_pressure":{"project_id":92,"device_code":"UF-4001A","plc_device_id":"0","item":"C.M.UF1_PT_JS@out","default_val":0},"product_pressure":{"project_id":92,"device_code":"UF-4001A","plc_device_id":"0","item":"C.M.UF1_PT_CS@out","default_val":0},"feed_wq_turbidity":{"project_id":92,"device_code":"AIT-94061","plc_device_id":"0","item":"C.M.UF_Tur_ZJS@out","default_val":0},"water_temperature":{"project_id":92,"device_code":"TIT-34061","plc_device_id":"0","item":"C.M.RO_TT_ZJS@out","default_val":0},"product_wq_ph":{"project_id":92,"device_code":"AIT-95065b","plc_device_id":"0","item":"C.M.UF_PH_ZCS@out","default_val":0},"step":{"project_id":92,"device_code":"HB","plc_device_id":"0","item":"C.M.UF1_DB@word_control","default_val":0}}`
 	var multiEnvItems MultiEnvItem
 
@@ -84,10 +94,40 @@ func TestMultiEnvItem_FillCurrentValue2(t *testing.T) {
 
 }
 
-func checkEarly(ft, fst float64, step string) bool {
-	return step == "26" && ft >= 360 && ft <= 660
+func TestEnvItem_Adjust(t *testing.T) {
+	cache := redis.NewClient(&redis.Options{Addr: "47.96.12.136:6379", Password: "", MaxRetries: 5})
+
+	SetOptions(Options{GtServerIp: "47.96.12.136:8788", Cache: cache, AdjustValue: true})
+	e := EnvItem{
+		ProjectId: 92,
+		Item:      "C.M.LT_QSC@out",
+	}
+
+	//v, ht, err := e.getCurrentValue()
+
+	//e.IncreAdjust(3600 * time.Second)
+	//e.IncreAdjust(3600 * time.Second)
+	e.IncreAdjust(3600 * time.Second)
+	//
+	//e.SetAdjust("65325", 3*time.Second)
+
+	adjust, err := e.GetAdjustInt64Val()
+
+	//t.Log(v, ht, err)
+
+	t.Log(adjust, err)
 }
 
-func checkAdvanced(ft, fst float64, step string) bool {
-	return step == "26" && ft+360 >= fst && ft+60 <= fst
+func TestEnvItem_GetValueForAdjust(t *testing.T) {
+	cache := redis.NewClient(&redis.Options{Addr: "47.96.12.136:6379", Password: "", MaxRetries: 5})
+
+	SetOptions(Options{GtServerIp: "47.96.12.136:8788", Cache: cache, AdjustValue: false})
+	e := EnvItem{
+		ProjectId: 92,
+		Item:      "C.M.LT_QSC@out",
+	}
+
+	v := e.GetItemFloat64Val()
+
+	t.Log(v)
 }

envitem/types.go

@@ -4,8 +4,9 @@ import "github.com/go-redis/redis/v8"
 
 type (
 	Options struct {
-		GtServerIp string
-		Cache      *redis.Client
+		GtServerIp  string
+		Cache       *redis.Client
+		AdjustValue bool
 		// 批量获取点位值
 		FetchMultiItem bool
 	}

envitem/vars.go

@@ -8,6 +8,9 @@ import (
 const (
 	CACHE_PREV_VALUE_KEY        = "envitem:prev:value:%d:%s"
 	CACHE_PREV_VALUE_KEY_EXPIRE = 86400 * time.Second
+
+	// CACHE_ADJUST_VALUE_KEY 会出现自控系统中点位值阻塞, 不发生变化的情况, 这里增加一个adjust来校准. 非常规操作
+	CACHE_ADJUST_VALUE_KEY = "envitem:adjust:value:%d:%s"
 )
 
 var (
@@ -15,4 +18,5 @@ var (
 	ctlUrl         = ""
 	cache          *redis.Client
 	fetchMultiItem = false
+	adjustValue    = false
 )

identify/multi_test.go

@@ -25,3 +25,44 @@ func TestNewMultiCheck(t *testing.T) {
 
 	t.Log(eventName)
 }
+
+func TestNewMultiCheck1(t *testing.T) {
+	//map[string]MultiRule{}
+	// ``
+	//ruleString := `{"过滤初期":{"C.M.UF1_DB@word_control":{"NvIn":["26"]},"C.M.UF1_DB@time_production":{"NvIn":["$C.M.UF1_DB@time_sp-360...$C.M.UF1_DB@time_sp-60"]},"C.M.UF1_DB@time_sp":{}}}`
+	ruleString := `{"xxx":{"C.M.UF4_DB@word_control":{},"C.M.UF4_DB@time_production_display":{},"C.M.UF4_DB@time_production":{},"C.M.UF4_DB@cycle_out":{"NvIn":["0"],"OvIn":["$C.M.UF4_DB@cycle_sp-1"]},"C.M.UF4_DB@cycle_sp":{}}}`
+
+	ovs := map[string]string{
+		"C.M.UF4_DB@word_control":            "25",
+		"C.M.UF4_DB@time_production":         "361",
+		"C.M.UF4_DB@time_production_display": "9632",
+		"C.M.UF4_DB@time_sp":                 "4500",
+		"C.M.UF4_DB@cycle_out":               "29",
+		"C.M.UF4_DB@cycle_sp":                "30",
+	}
+
+	nvs := map[string]string{
+		"C.M.UF4_DB@word_control":            "26",
+		"C.M.UF4_DB@time_production_display": "9632",
+		"C.M.UF4_DB@time_production":         "8839",
+		"C.M.UF4_DB@time_sp":                 "9000",
+		"C.M.UF4_DB@cycle_out":               "0",
+		"C.M.UF4_DB@cycle_sp":                "33",
+	}
+
+	eventName := MultiTest(ovs, nvs, ruleString)
+
+	t.Log(eventName)
+}
+
+func TestNewMultiCheck2(t *testing.T) {
+	x := map[string]interface{}{
+		"x":    1,
+		"xb":   "b",
+		"asdf": nil,
+	}
+
+	for s, i := range x {
+		t.Logf("name: %s %#v\n", s, i)
+	}
+}

identify/single_test.go

@@ -5,7 +5,7 @@ import "testing"
 func TestStepTest(t *testing.T) {
 	// 只要值发生改变就触发  旧值的可选范围中, 不包含新值
 	// 适合液位变化, 开关量, 敏感数据监控
-	rule := `{"xxx":{"OvNotIn":["$nv"]}}`
-	x := StepTest("5", "6", rule)
+	rule := `{"xxx":{"NvIn":["1"],"OvIn":["0"]}}`
+	x := StepTest("0", "0", rule)
 	t.Log(x)
 }

rabbitMQ/producter_test.go

@@ -17,7 +17,7 @@ func TestNewProducter(t *testing.T) {
 
 	t.Log(got, err)
 
-	send := got.Publisher("step_change", []byte(`{"device_code":"UF_4001A"}`))
+	send := got.Publisher("EVENT:NOFITY:92:CEB_TRIGGER", []byte(`{"device_code":"UF_4001A"}`))
 
 	t.Log(send)