water-system-gateway/server/client.go

package server

import (
	"context"
	"errors"
	"fmt"
	"github.com/gogf/gf/v2/encoding/gbinary"
	"github.com/gogf/gf/v2/net/gtcp"
	"github.com/gogf/gf/v2/os/glog"
	"github.com/gogf/gf/v2/util/gconv"
	"io"
	"math"
	"net"
	"strings"
	"syscall"
	"time"
	"water-system-gateway/protocol"
)

type Client struct {
	Id           string
	ctx          context.Context
	srv          *Server
	conn         *gtcp.Conn
	sendChan     chan []byte
	closeChan    chan struct{}
	closeHandler func(id string, c *Client)
	regHandler   func(id string, c *Client)
	isReg        bool
}

func NewClient(s *Server, conn *gtcp.Conn) *Client {
	return &Client{
		srv:       s,
		ctx:       context.Background(),
		conn:      conn,
		sendChan:  make(chan []byte),
		closeChan: make(chan struct{}),
	}
}

func (c *Client) SetId(id string) {
	c.Id = id
}

func (c *Client) SendLoop() {
	for {
		select {
		case buf := <-c.sendChan:
			err := c.send(buf)
			if err != nil {
				glog.Errorf(c.ctx, "指令发送失败:%s", err.Error())
				continue
			}
			for {
				select {
				case <-c.closeChan:
					return
				case <-time.After(2 * time.Second):
					glog.Error(c.ctx, "接收指令超时")
					break
				default:
					receiveBuf, err := c.conn.Recv(-1)
					if err != nil {
						c.readError(err)
						break
					}
					if !c.isReg {
						fmt.Println(receiveBuf)
						id := gbinary.DecodeToString(receiveBuf)
						glog.Debugf(c.ctx, "收到注册包!id:%s", id)
						c.SetId(id)
						c.isReg = true
						if c.regHandler != nil {
							c.regHandler(c.Id, c)
						}
						continue
					}
					glog.Debugf(c.ctx, "收到数据：%2X", receiveBuf)
					if err := c.decodeAndReport(receiveBuf); err != nil {
						glog.Debugf(c.ctx, "处理数据失败:%s", err.Error())
						break
					}
				}
				break
			}
		}
	}
}

// 收到数据：01 03 04 00 FD 00 8A EA 64
func (c *Client) decodeAndReport(buf []byte) error {
	//length := len(buf)
	//var crc crc
	//crc.reset().pushBytes(buf[0 : length-2])
	//checksum := uint16(buf[length-1])<<8 | uint16(buf[length-2])
	//if checksum != crc.value() {
	//	return errors.New(fmt.Sprintf("modbus: response crc '%v' does not match expected '%v'", checksum, crc.value()))
	//}
	if len(buf) <= 8 {
		return nil
	}
	if buf[1] == 0x03 {
		if buf[2] == 0xBE {
			data := &protocol.WaterSystem{}
			data.Power1 = gconv.Int(gbinary.BeDecodeToUint16(buf[3:5]))
			data.Power2 = gconv.Int(gbinary.BeDecodeToUint16(buf[5:7]))
			data.Power3 = gconv.Int(gbinary.BeDecodeToUint16(buf[7:9]))
			data.Power4 = gconv.Int(gbinary.BeDecodeToUint16(buf[9:11]))
			data.Power5 = gconv.Int(gbinary.BeDecodeToUint16(buf[11:13]))
			data.Power6 = gconv.Int(gbinary.BeDecodeToUint16(buf[13:15]))
			data.Power7 = gconv.Int(gbinary.BeDecodeToUint16(buf[15:17]))
			data.Power8 = gconv.Int(gbinary.BeDecodeToUint16(buf[17:19]))
			data.HeaderPower = gconv.Int(gbinary.BeDecodeToUint16(buf[21:23]))
			data.InletPressure = gconv.Float32(gbinary.BeDecodeToUint16(buf[23:25])) / 10
			data.OutletPressure = gconv.Float32(gbinary.BeDecodeToUint16(buf[25:27])) / 10
			data.PressureDifference = gconv.Float32(gbinary.BeDecodeToUint16(buf[27:29])) / 10
			data.FaultCode = gconv.Int(gbinary.BeDecodeToUint16(buf[29:31]))
			data.FillValve = gconv.Int(gbinary.BeDecodeToUint16(buf[31:33]))
			data.BypassValve = gconv.Int(gbinary.BeDecodeToUint16(buf[33:35]))
			data.SetInletWaterPre = gconv.Float32(gbinary.BeDecodeToUint16(buf[35:37])) / 10
			data.SetMaxDifferencePre = gconv.Float32(gbinary.BeDecodeToUint16(buf[37:39])) / 10
			data.SetMinDifferencePre = gconv.Float32(gbinary.BeDecodeToUint16(buf[39:41])) / 10
			data.HeaderMode = gconv.Int(gbinary.BeDecodeToUint16(buf[41:43]))

			data.PrimaryPumpStatus = gconv.Int(gbinary.BeDecodeToUint16(buf[43:45]))
			data.SecondaryPumpStatus = gconv.Int(gbinary.BeDecodeToUint16(buf[45:47]))
			data.PrimaryPumpFrequency = gconv.Float32(gbinary.BeDecodeToUint16(buf[47:49])) / 10
			data.SecondaryPumpFrequency = gconv.Float32(gbinary.BeDecodeToUint16(buf[49:51])) / 10
			data.WetNodeInput1 = gconv.Int(gbinary.BeDecodeToUint16(buf[51:53]))
			data.WetNodeInput2 = gconv.Int(gbinary.BeDecodeToUint16(buf[53:55]))
			data.WetNodeInput = gconv.Int(gbinary.BeDecodeToUint16(buf[55:57]))
			data.DryNodeInput = gconv.Int(gbinary.BeDecodeToUint16(buf[57:59]))
			data.PrimaryOutletTemp = gconv.Float32(gbinary.BeDecodeToUint16(buf[59:61])) / 10
			data.PrimaryReturnTemp = gconv.Float32(gbinary.BeDecodeToUint16(buf[61:63])) / 10
			data.SecondaryOutletTemp = gconv.Float32(gbinary.BeDecodeToUint16(buf[63:65])) / 10
			data.SecondaryReturnTemp = gconv.Float32(gbinary.BeDecodeToUint16(buf[65:67])) / 10

			data.CollectorPower = gconv.Int(gbinary.BeDecodeToUint16(buf[81:83]))
			data.PumpStationPower = gconv.Int(gbinary.BeDecodeToUint16(buf[83:85]))
			data.DeviceReset = gconv.Int(gbinary.BeDecodeToUint16(buf[85:87]))
			data.CollectorMode = gconv.Int(gbinary.BeDecodeToUint16(buf[87:89]))
			data.SetTargetOutletPressure = gconv.Float32(gbinary.BeDecodeToUint16(buf[89:91])) / 10
			data.SetMaxiPressureDifference = gconv.Float32(gbinary.BeDecodeToUint16(buf[91:93])) / 10
			data.SetMiniPressureDifference = gconv.Float32(gbinary.BeDecodeToUint16(buf[93:95])) / 10
			data.SetBypassPipeDiameter = gconv.Int(gbinary.BeDecodeToUint16(buf[95:97]))
			data.FillValvePower = gconv.Int(gbinary.BeDecodeToUint16(buf[97:99]))
			data.PrimaryPumpTypeConf = gconv.Int(gbinary.BeDecodeToUint16(buf[99:101]))
			data.SecondaryPumpTypeConf = gconv.Int(gbinary.BeDecodeToUint16(buf[101:103]))
			data.PrimaryPumpControlMethod = gconv.Int(gbinary.BeDecodeToUint16(buf[103:105]))
			data.SecondaryPumpControlMethod = gconv.Int(gbinary.BeDecodeToUint16(buf[105:107]))
			data.SetPrimaryPumpConstantPressure = gconv.Float32(gbinary.BeDecodeToUint16(buf[107:109])) / 10
			data.SetSecondaryPumpConstantPressure = gconv.Float32(gbinary.BeDecodeToUint16(buf[109:111])) / 10
			data.SetPrimaryPumpConstantTempDifference = gconv.Float32(gbinary.BeDecodeToUint16(buf[111:113])) / 10
			data.SetSecondaryPumpConstantTempDifference = gconv.Float32(gbinary.BeDecodeToUint16(buf[113:115])) / 10
			data.PrimaryPumpFrequencyLowerLimit = gconv.Float32(gbinary.BeDecodeToUint16(buf[115:117]))
			data.PrimaryPumpFrequencyUpperLimit = gconv.Float32(gconv.Float32(gbinary.BeDecodeToUint16(buf[117:119])))
			data.PumpPWMControlTypeConf = gconv.Int(gbinary.BeDecodeToUint16(buf[119:121]))
			data.SecondaryPumpFrequencyLowerLimit = gconv.Float32(gbinary.BeDecodeToUint16(buf[121:123]))
			data.SecondaryPumpFrequencyUpperLimit = gconv.Float32(gbinary.BeDecodeToUint16(buf[123:125]))
			data.AntiJammingFunction = gconv.Int(gbinary.BeDecodeToUint16(buf[125:127]))
			data.BypassValvePower = gconv.Int(gbinary.BeDecodeToUint16(buf[127:129]))
			data.OperatingMode = gconv.Int(gbinary.BeDecodeToUint16(buf[129:131]))

			data.OutdoorMode = gconv.Int(gbinary.BeDecodeToUint16(buf[173:175]))
			data.RefrigerationWaterTemp = gconv.Float32(gbinary.BeDecodeToUint16(buf[175:177])) / 10
			data.HeatingWaterTemp = gconv.Float32(gbinary.BeDecodeToUint16(buf[177:179])) / 10
			data.OutdoorCirculationTemp = gconv.Float32(gbinary.BeDecodeToUint16(buf[151:153]))
			data.InletWaterTemp = gconv.Float32(gbinary.BeDecodeToUint16(buf[153:155]))
			data.OutletWaterTemp = gconv.Float32(gbinary.BeDecodeToUint16(buf[155:157]))
			data.OutdoorPower = gconv.Int(gbinary.BeDecodeToUint16(buf[179:181]))
			data.Auxiliary = gconv.Int(gbinary.BeDecodeToUint16(buf[159:161]))
			data.FaultMark = gconv.Int(gbinary.BeDecodeToUint16(buf[161:163]))

			if err := c.srv.ReportStatus(c.Id, data, "status"); err != nil {
				return err
			}
		}
	}

	return nil
}

func (c *Client) readError(err error) {
	defer c.closeConnection()
	if err == io.EOF || isErrConnReset(err) {
		return
	}
	glog.Errorf(c.ctx, "读取数据发生错误：%s", err.Error())
}

func (c *Client) closeConnection() {
	_ = c.conn.Close()
	c.conn = nil
	close(c.closeChan)
	c.isReg = false
	if c.closeHandler != nil {
		c.closeHandler(c.Id, c)
	}
}

// 计算温度值, 处理零下的情况
func caleTemperature(data []byte) float32 {
	value := gconv.Float32(gbinary.BeDecodeToUint16(data)&0x7FFF) / 10
	if value < 1 {
		value = float32(math.Floor(float64(value) + 0.5))
	}
	if gbinary.BeDecodeToUint16(data)&0x8000 == 0x8000 && value != 0 {
		value = value * -1
	}
	return value
}

// isErrConnReset read: connection reset by peer
func isErrConnReset(err error) bool {
	var ne *net.OpError
	if errors.As(err, &ne) {
		return strings.Contains(ne.Err.Error(), syscall.ECONNRESET.Error())
	}
	return false
}

//func dataBlock(value float32) []byte {
//	buffer := &bytes.Buffer{}
//	var a uint16
//	if value >= 0 {
//		a = uint16(value*100) | 0x8000
//	} else {
//		a = uint16(math.Abs(float64(value))*100) | 0x0000
//	}
//	buffer.Write(gbinary.BeEncodeUint16(a))
//	return buffer.Bytes()
//}

func (c *Client) GetSendByte() {
	for {
		c.sendChan <- []byte{0x01, 0x03, 0x00, 0x00, 0x00, 0x5F, 0x05, 0xF2}
		time.Sleep(10 * time.Second)
	}
}

func (c *Client) send(buf []byte) error {
	if c.conn == nil {
		return nil
	}
	glog.Debugf(c.ctx, "----->%2X", buf)
	err := c.conn.Send(buf)
	if err != nil {
		glog.Error(c.srv.ctx, err)
		c.closeConnection()
		return err
	}
	return nil
}

func (c *Client) WaterSystemControl(address uint16, value interface{}) error {
	result, err := WriteMultipleRegisters(address, 1, gbinary.BeEncodeUint16(gconv.Uint16(value)))
	if err != nil {
		return err
	}
	c.sendChan <- result
	time.Sleep(100 * time.Millisecond)
	c.sendChan <- []byte{0x01, 0x03, 0x00, 0x00, 0x00, 0x5F, 0x05, 0xF2}
	return nil
}