apache plc4x短连接

zhangqian

2023-09-16 d457b47a702857cdefb56a0a518201ba8ad2505f

apache plc4x短连接

 api/v1/plc.go

@@ -6,7 +6,7 @@
    "apsClient/model/response"
    _ "apsClient/model/response"
    "apsClient/pkg/contextx"
    "apsClient/pkg/plc/modbusx"
    "apsClient/pkg/plc/apacheplc4x"
    "apsClient/service"
    "github.com/gin-gonic/gin"
    "github.com/spf13/cast"
@@ -34,7 +34,7 @@
    resp.TotalNumber = cast.ToInt(totalNumber)

    plcStatus := 1 //断开连接
    isConnect := modbusx.IsConnect()
    isConnect := apacheplc4x.IsConnect()
    if isConnect {
        if resp.FinishNumber > 0 { //生产
            plcStatus = 2

 pkg/plc/apacheplc4x/modbus.go

New file
@@ -0,0 +1,203 @@
package apacheplc4x

import (
    "apsClient/pkg/logx"
    "context"
    "errors"
    "fmt"
    plc4go "github.com/apache/plc4x/plc4go/pkg/api"
    "github.com/apache/plc4x/plc4go/pkg/api/drivers"
    apiModel "github.com/apache/plc4x/plc4go/pkg/api/model"
    "github.com/apache/plc4x/plc4go/pkg/api/transports"
    "sync/atomic"
    "time"
)

func GetModbusConnection(ipAddr string) (plc4go.PlcConnection, error) {
    // 创建一个上下文，并设置 3 秒超时
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()
    conn, err := newGetModbusConnection(ctx, ipAddr)
    if err != nil {
        logx.Errorf("new modbus connection err: %v", err.Error())
        return nil, err
    }
    return conn, nil
}

func newGetModbusConnection(ctx context.Context, ipAddr string) (plc4go.PlcConnection, error) {
    // 创建驱动管理器
    driverManager := plc4go.NewPlcDriverManager()
    // 注册TCP传输
    transports.RegisterTcpTransport(driverManager)
    // 注册驱动
    //drivers.RegisterKnxDriver(driverManager)
    drivers.RegisterModbusTcpDriver(driverManager)

    // 通过TCP连接PLC设备
    connectionString := fmt.Sprintf("modbus-tcp://%s", ipAddr)
    connectionRequestChanel := driverManager.GetConnection(connectionString)

    // 等待连接响应，同时考虑上下文的超时
    select {
    case connectionResult := <-connectionRequestChanel:
        if err := connectionResult.GetErr(); err != nil {
            return nil, err
        }
        return connectionResult.GetConnection(), nil
    case <-ctx.Done():
        return nil, ctx.Err()
    }
}

func readHoldingRegisterSingle(connection plc4go.PlcConnection, address int) ([]byte, error) {
    tag := fmt.Sprintf("tag:%v", address)
    tagAddress := fmt.Sprintf("holding-register:%d:UINT", address)

    // 读模式
    readRequest, err := connection.ReadRequestBuilder().AddTagAddress(tag, tagAddress).Build()
    if err != nil {
        logx.Errorf("preparing read-request:%s\n", err.Error())
        return nil, err
    }

    // 执行
    readResult := <-readRequest.Execute()
    if err := readResult.GetErr(); err != nil {
        logx.Errorf("execting read-request:%s\n", err.Error())
        return nil, err
    }

    // 判断响应码是否正确
    if readResult.GetResponse().GetResponseCode(tag) != apiModel.PlcResponseCode_OK {
        logx.Errorf("error an non-ok return code: %s", readResult.GetResponse().GetResponseCode(tag).GetName())
        return nil, nil
    }

    value := readResult.GetResponse().GetValue(tag)

    return value.GetRaw(), err

}

func readHoldingRegisterList(connection plc4go.PlcConnection, address, length int) ([]byte, error) {
    tag := fmt.Sprintf("tag:%v:%v", address, length)
    tagAddress := fmt.Sprintf("holding-register:%d:UINT[%d]", address, length)

    // 读模式
    readRequest, err := connection.ReadRequestBuilder().AddTagAddress(tag, tagAddress).Build()
    if err != nil {
        logx.Errorf("preparing read-request:%s\n", err.Error())
        return nil, err
    }

    // 执行
    readResult := <-readRequest.Execute()
    if err := readResult.GetErr(); err != nil {
        logx.Errorf("execting read-request:%s\n", err.Error())
        return nil, err
    }

    // 判断响应码是否正确
    if readResult.GetResponse().GetResponseCode(tag) != apiModel.PlcResponseCode_OK {
        logx.Errorf("error an non-ok return code: %s", readResult.GetResponse().GetResponseCode(tag).GetName())
        return nil, errors.New("error  code: " + readResult.GetResponse().GetResponseCode(tag).GetName())
    }

    value := readResult.GetResponse().GetValue(tag)

    var result []byte

    for _, val := range value.GetList() {
        result = append(result, val.GetRaw()...)
    }

    return result, nil
}

func ReadHoldingRegister(ipAddr string, address, length int) ([]byte, error) {
    connection, err := GetModbusConnection(ipAddr)
    if err != nil {
        dealErr(err, ipAddr)
        return nil, err
    }
    defer connection.Close()
    if length > 1 {
        return readHoldingRegisterList(connection, address, length)
    }

    return readHoldingRegisterSingle(connection, address)
}

func WriteHoldingRegister(ipAddr string, address int, value any) (string, error) {
    connection, err := GetModbusConnection(ipAddr)
    if err != nil {
        dealErr(err, ipAddr)
        return "", err
    }
    defer connection.Close()
    tag := fmt.Sprintf("tag:%v:w", address)
    tagAddress := fmt.Sprintf("holding-register:%d:UINT", address)

    // 写模式
    writeRequest, err := connection.WriteRequestBuilder().AddTagAddress(tag, tagAddress, value).Build()
    if err != nil {
        logx.Errorf("preparing read-request:%s\n", err.Error())
        return "", err
    }

    // 执行
    writeResult := <-writeRequest.Execute()
    if err := writeResult.GetErr(); err != nil {
        logx.Errorf("execting read-request:%s\n", err.Error())
        return "", err
    }

    // 判断响应码是否正确
    if writeResult.GetResponse().GetResponseCode(tag) != apiModel.PlcResponseCode_OK {
        logx.Errorf("error an non-ok return code: %s", writeResult.GetResponse().GetResponseCode(tag).GetName())
        return "", errors.New("error  code: " + writeResult.GetResponse().GetResponseCode(tag).GetName())
    }

    result := writeResult.GetResponse().String()

    return result, nil
}

func dealErr(err error, ipAddr string) {
    if err != nil {
        FailureRemainingOpportunitiesDecr() //减少失败剩余机会
    } else {
        FailureRemainingOpportunitiesReset() //重置失败剩余机会
    }
}

var connectionStatus atomic.Bool

var failureRemainingOpportunities atomic.Int64

const (
    defaultFailureRemainingOpportunities = 20
)

func IsConnect() bool {
    return connectionStatus.Load()
}

func FailureRemainingOpportunitiesDecr() {
    newValue := failureRemainingOpportunities.Add(-1)
    if newValue <= 0 {
        connectionStatus.Store(false)
    }
    return
}

func FailureRemainingOpportunitiesReset() {
    if failureRemainingOpportunities.Load() < defaultFailureRemainingOpportunities {
        failureRemainingOpportunities.Store(defaultFailureRemainingOpportunities)
    }
    if connectionStatus.Load() == false {
        connectionStatus.Store(true)
    }
    return
}

 service/plc.go

@@ -5,67 +5,13 @@
    "apsClient/constvar"
    "apsClient/model"
    "apsClient/pkg/logx"
    "apsClient/pkg/plc"
    "apsClient/pkg/plc/modbusx"
    "apsClient/pkg/plc/apacheplc4x"
    "apsClient/pkg/plccom"
    "encoding/binary"
    "errors"
    "fmt"
    "github.com/spf13/cast"
)

func PlcRead(plcConfig *model.DevicePlc, fieldType constvar.PlcStartAddressType, position int) (val interface{}, err error) {
    var (
        startAddress int
        valueType    constvar.PlcStartAddressValueType
        dataLength   int
        ipAddr       string
    )

    for _, pc := range plcConfig.Details {
        if pc.FieldName == fieldType && pc.Position == position {
            startAddress = pc.StartAddress
            valueType = pc.Type
            dataLength = pc.Length
        }
    }

    if plcConfig.Method == constvar.PlcMethodModbusTCP {
        ipAddr = fmt.Sprintf("%s:%v", plcConfig.Address, plcConfig.Port)

        conn, err := plc.GetModbusConnection(ipAddr)
        if err != nil {
            logx.Errorf("PlcRead 连接plc失败: %v", err.Error())
            return nil, err
        }

        rawData, err := plc.ReadHoldingRegister(conn, startAddress, dataLength)
        if err != nil {
            logx.Errorf("PlcRead 获取plc数据失败: %v", err.Error())
            return nil, err
        }
        switch valueType {
        case constvar.PlcStartAddressValueTypeString:
            return string(rawData), nil
        case constvar.PlcStartAddressValueTypeInt:
            if len(rawData) == 2 {
                return int(binary.BigEndian.Uint16(rawData)), nil
            } else {
                logx.Errorf("plc read get an unknown int value: %v, address:%v", rawData, startAddress)
                return nil, errors.New(fmt.Sprintf("unknown int value：%v", rawData))
            }
        }
        return nil, errors.New("undefined value type")
    } else if plcConfig.Method == constvar.PlcMethodSerial {
        ipAddr = conf.Conf.Services.Serial
        if ipAddr == "" {
            return nil, errors.New("conf.Conf.Services.Serial config not set yet")
        }
        label := fmt.Sprintf("D%d", startAddress)
        return plccom.ReadPLC(plccom.DeviceTypeMitsubishi, ipAddr, label)
    }
    return nil, errors.New("interface type not support")
}

func PlcWrite(plcConfig *model.DevicePlc, fieldType constvar.PlcStartAddressType, position int, value interface{}) (err error) {
    var (
@@ -94,7 +40,11 @@
        //}
        //
        //result, err := plc.WriteHoldingRegister(conn, startAddress, value)
        err = modbusx.Write(ipAddr, uint16(startAddress), value)

        //err = modbusx.Write(ipAddr, uint16(startAddress), value)

        _, err = apacheplc4x.WriteHoldingRegister(ipAddr, startAddress, value)

        if err != nil {
            logx.Errorf("plc write failed, address: %v, value: %v, err: %v", startAddress, value, err.Error())
            plcConfig.CurrentErr = err
@@ -126,7 +76,10 @@
        //}
        //result, err := plc.WriteHoldingRegister(conn, address, value)

        err = modbusx.Write(ipAddr, uint16(address), value)
        //err = modbusx.Write(ipAddr, uint16(address), value)

        _, err = apacheplc4x.WriteHoldingRegister(ipAddr, address, value)

        if err != nil {
            logx.Errorf("plc write failed, address: %v, value: %v, err: %v", address, value, err.Error())
            return err
@@ -155,7 +108,11 @@
        //    return PlcReadDirect(plcConfig, address, dataLength, valueType)
        //}
        //value, err := plc.ReadHoldingRegister(conn, address, dataLength)
        value, err := modbusx.Read(ipAddr, uint16(address), uint16(dataLength))

        //value, err := modbusx.Read(ipAddr, uint16(address), uint16(dataLength))

        value, err := apacheplc4x.ReadHoldingRegister(ipAddr, address, dataLength)

        if err != nil {
            return nil, err
        }