gt-common/envitem/func.go

package envitem

import (
	"context"
	"database/sql/driver"
	"encoding/json"
	"errors"
	"fmt"
	"metawant.greentech.com.cn/gaoyagang/gt-common/httplib"
	"strconv"
	"strings"
	"sync"
	"time"
)

func SetOptions(options Options) {
	snapUrl = fmt.Sprintf("http://%s/api/v1/plc-current", options.GtServerIp)
	ctlUrl = fmt.Sprintf("http://%s/api/v1/plc/set-var-values", options.GtServerIp)
	plcItemUrl = fmt.Sprintf("http://%s/api/v1/plc-list/", options.GtServerIp)

	if options.Cache != nil {
		cache = options.Cache
	}

	fetchMultiItem = options.FetchMultiItem
	adjustValue = options.AdjustValue
}

func (m MultiEnvItem) GetProjectId() int64 {
	for _, item := range m {
		return item.ProjectId
	}
	return 0
}

func (m MultiEnvItem) getItemNames() []string {
	items := make([]string, 0)
	for _, item := range m {
		items = append(items, item.Item)
	}
	return items
}

func (m MultiEnvItem) FillCurrentValue() error {
	datas := make([]*ItemValueResp, 0)
	var err error
	if fetchMultiItem {
		datas, err = m.getCurrentData()
	} else {
		datas, err = m.getCurrentDataByOne()
	}

	if err != nil {
		return err
	}
	// 先做一个item -> key的映射
	ikm := make(map[string]string)
	for s, item := range m {
		ikm[item.Item] = s
	}

	for _, data := range datas {
		if data != nil {
			if k, ok := ikm[data.ItemName]; ok {
				m[k].Value = data.Val
				m[k].Htime = data.HTime
				_ = m[k].setPrevValue()
			}
		}
	}
	return nil
}

func (m MultiEnvItem) GetItemFloat64Value(key string) float64 {
	if envItem, ok := m[key]; ok {
		return envItem.GetItemFloat64Val()
	}
	return 0
}

func (m MultiEnvItem) GetItemInt64Value(key string) int64 {
	if envItem, ok := m[key]; ok {
		return envItem.GetItemInt64Val()
	}
	return 0
}

func (m MultiEnvItem) GetItemStringValue(key string) string {
	if envItem, ok := m[key]; ok {
		return envItem.GetItemStringVal()
	}
	return ""
}

func (m MultiEnvItem) GetItemHtime(key string) *time.Time {
	if envItem, ok := m[key]; ok {
		return envItem.GetItemHtime()
	}
	return nil
}

func (m MultiEnvItem) FindString() map[string]string {
	a := make(map[string]string)
	for _, item := range m {
		a[item.Item] = item.GetItemStringVal()
	}

	return a
}

func (m MultiEnvItem) FindPrevString() map[string]string {
	a := make(map[string]string)
	for _, item := range m {
		a[item.Item] = item.GetItemPrevStringVal()
	}

	return a
}

func (m MultiEnvItem) ClearValues() {
	for _, item := range m {
		item.clearValue()
	}
}

func (m MultiEnvItem) getCurrentDataByOne() ([]*ItemValueResp, error) {
	datas := make([]*ItemValueResp, len(m))
	i := 0
	var wg sync.WaitGroup
	wg.Add(len(m))
	for _, item := range m {
		go func(index int, one *EnvItem) {
			defer wg.Done()
			datas[index], _ = one.getCurrentData()
		}(i, item)
		i++
	}
	wg.Wait()
	return datas, nil
}

func (m MultiEnvItem) getCurrentData() ([]*ItemValueResp, error) {
	req := httplib.Post(snapUrl)
	data := make([]*ItemValueReq, 1)

	data[0] = &ItemValueReq{
		DeviceItems: strings.Join(m.getItemNames(), ","),
		ProjectId:   m.GetProjectId(),
	}

	jsonBytes, err := json.Marshal(data)
	if err != nil {
		return nil, err
	}
	req.Body(jsonBytes)
	req.SetTimeout(time.Millisecond * 2000)

	r, err := req.Response()
	if err != nil {
		return nil, err
	}
	defer r.Body.Close()

	if r.StatusCode == 200 {
		resp, err := req.Bytes()
		if err != nil {
			return nil, err
		}

		res := &ItemValueResps{}
		err = json.Unmarshal(resp, res)
		if err != nil {
			return nil, err
		}

		if len(res.Data) == 0 {
			return nil, errors.New("not found envitem's value")
		}

		return res.Data, nil
	}

	return nil, errors.New(fmt.Sprintf("request statusCode: %d", r.StatusCode))
}

func (e *EnvItem) getCurrentData() (*ItemValueResp, error) {
	req := httplib.Post(snapUrl)
	data := make([]*ItemValueReq, 1)

	data[0] = &ItemValueReq{
		DeviceItems: e.Item,
		ProjectId:   e.ProjectId,
	}

	jsonBytes, err := json.Marshal(data)
	if err != nil {
		return nil, err
	}
	req.Body(jsonBytes)
	req.SetTimeout(time.Millisecond * 2000)

	r, err := req.Response()
	if err != nil {
		return nil, err
	}
	defer r.Body.Close()

	if r.StatusCode == 200 {
		resp, err := req.Bytes()
		if err != nil {
			return nil, err
		}

		res := &ItemValueResps{}
		err = json.Unmarshal(resp, res)
		if err != nil {
			return nil, err
		}

		if len(res.Data) == 0 {
			return nil, errors.New("not found envitem's value")
		}

		if adjustValue && cache != nil && len(res.Data) > 0 {
			if rv, err := strconv.ParseInt(res.Data[0].Val, 10, 64); err == nil {
				adjust, _ := e.GetAdjustInt64Val()
				res.Data[0].Val = fmt.Sprintf("%d", rv+adjust)
			} else 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
	}

	return nil, errors.New(fmt.Sprintf("request statusCode: %d", r.StatusCode))
}

func (e *EnvItem) getCurrentValue() (string, string, error) {
	resp, err := e.getCurrentData()
	if err != nil {
		return "", "", err
	}

	e.Value = resp.Val
	e.Htime = resp.HTime
	_ = e.setPrevValue()

	return resp.Val, resp.HTime, nil
}

func (e *EnvItem) getPrevValue() (string, string, error) {
	if cache == nil {
		return "", "", errors.New("not cache")
	}
	key := fmt.Sprintf(CACHE_PREV_VALUE_KEY, e.ProjectId, e.Item)
	if cmd := cache.Get(context.Background(), key); cmd != nil {
		n := strings.Split(cmd.Val(), "|")
		if len(n) != 2 {
			return "", "", errors.New("getPrevValue cache length error")
		}
		return n[0], n[1], nil
	} else {
		return "", "", cmd.Err()
	}
}

func (e *EnvItem) setPrevValue() error {
	if cache == nil {
		return errors.New("not cache")
	}
	key := fmt.Sprintf(CACHE_PREV_VALUE_KEY, e.ProjectId, e.Item)
	value := fmt.Sprintf("%s|%s", e.Value, e.Htime)
	if statusCmd := cache.Set(context.Background(), key, value, CACHE_PREV_VALUE_KEY_EXPIRE); statusCmd != nil {
		return nil
	} else {
		return errors.New("setPrevValue error")
	}
}

func (e *EnvItem) GetItemFloat64Val() float64 {
	if e.Value == nil {
		e.getCurrentValue()
	}

	switch e.Value.(type) {
	case string:
		if v, e := strconv.ParseFloat(e.Value.(string), 64); e == nil {
			return v
		}
	}

	return 0
}

func (e *EnvItem) GetItemInt64Val() int64 {
	if e.Value == nil {
		e.getCurrentValue()
	}

	switch e.Value.(type) {
	case string:
		if v, e := strconv.ParseInt(e.Value.(string), 10, 64); e == nil {
			return v
		}
	}

	return 0
}

func (e *EnvItem) GetItemStringVal() string {
	if e.Value == nil {
		e.getCurrentValue()
	}

	if v, ok := e.Value.(string); ok {
		return v
	}

	return ""
}

func (e *EnvItem) GetItemHtime() *time.Time {
	if e.Value == nil {
		e.getCurrentValue()
	}

	if ht, err := time.ParseInLocation("2006-01-02 15:04:05", e.Htime, time.Local); err == nil {
		return &ht
	}
	return nil
}

func (e *EnvItem) GetItemPrevFloat64Val() float64 {
	ov, _, err := e.getPrevValue()
	if err != nil {
		return 0
	}

	if v, err := strconv.ParseFloat(ov, 64); err == nil {
		return v
	}

	return 0
}

func (e *EnvItem) GetItemPrevInt64Val() int64 {
	ov, _, err := e.getPrevValue()
	if err != nil {
		return 0
	}

	if v, err := strconv.ParseInt(ov, 10, 64); err == nil {
		return v
	}

	return 0
}

func (e *EnvItem) GetItemPrevStringVal() string {
	ov, _, _ := e.getPrevValue()
	return ov
}

func (e *EnvItem) GetItemPrevHtime() *time.Time {
	_, ht, _ := e.getPrevValue()
	if t, err := time.ParseInLocation("2006-01-02 15:04:05", ht, time.Local); err == nil {
		return &t
	}
	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
}

// WaitValue 等待点位的值, 变为指定值
// v: 等待的值
// op: -1: 小于, 0: 等于, 1: 大于
// interval: 轮询间隔
// retry: 最多尝试次数 0: 表示不限制
func (e *EnvItem) WaitValue(v string, op int, interval time.Duration, retry int) <-chan bool {
	c := make(chan bool)
	go func() {
		t := 0
		for {
			val, _, err := e.getCurrentValue()
			if err == nil && strings.Compare(v, val) == op {
				c <- true
			}

			if retry > 0 {
				t += 1
				if t >= retry {
					c <- false
				}
			}

			time.Sleep(interval)
		}
	}()

	return c
}

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 = ""
}

// Scan 实现方法
func (d *MultiEnvItem) Scan(input interface{}) error {
	_ = json.Unmarshal(input.([]byte), &d)
	return nil
}

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
}

func (v *VirtualPlcItem) GetVirtualPlcItems() (*VirtualPlcItemResp, error) {

	tmpPlcItemUrl := fmt.Sprintf("%s%d", plcItemUrl, v.ProjectId)

	req := httplib.Get(tmpPlcItemUrl)
	req.Param("project_id", fmt.Sprintf("%d", v.ProjectId))
	req.Param("pageSize", fmt.Sprintf("%d", v.PageSize))

	r, err := req.Response()
	if err != nil {
		return nil, err
	}
	defer r.Body.Close()

	if r.StatusCode == 200 {
		resp, err := req.Bytes()
		if err != nil {
			return nil, err
		}

		res := &VirtualPlcItemResp{}
		err = json.Unmarshal(resp, res)
		if err != nil {
			return nil, err
		}
		return res, nil
	}

	return nil, errors.New(fmt.Sprintf("request statusCode: %d", r.StatusCode))
}