You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@plc4x.apache.org by sr...@apache.org on 2021/04/12 21:38:59 UTC
[plc4x] branch develop updated: plc4go: first draft of write in s7
This is an automated email from the ASF dual-hosted git repository.
sruehl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
The following commit(s) were added to refs/heads/develop by this push:
new 8004549 plc4go: first draft of write in s7
8004549 is described below
commit 80045494b056223ae727f5b77b9fa7ae7be45e51
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Mon Apr 12 23:38:47 2021 +0200
plc4go: first draft of write in s7
+ basic write implemented
- There seems to be a bug in the generated code see s7_driver_test.go
---
plc4go/internal/plc4go/ads/Writer.go | 1 +
plc4go/internal/plc4go/s7/Connection.go | 2 +-
plc4go/internal/plc4go/s7/Field.go | 29 +++
plc4go/internal/plc4go/s7/Reader.go | 27 +--
plc4go/internal/plc4go/s7/Writer.go | 314 ++++++++++++++++++--------------
5 files changed, 220 insertions(+), 153 deletions(-)
diff --git a/plc4go/internal/plc4go/ads/Writer.go b/plc4go/internal/plc4go/ads/Writer.go
index 5fdd18b..beebb71 100644
--- a/plc4go/internal/plc4go/ads/Writer.go
+++ b/plc4go/internal/plc4go/ads/Writer.go
@@ -143,6 +143,7 @@ func (m *Writer) Write(writeRequest model.PlcWriteRequest) <-chan model.PlcWrite
}
// Calculate a new unit identifier
+ // TODO: this is not threadsafe as the whole operation is not atomic
transactionIdentifier := atomic.AddUint32(&m.transactionIdentifier, 1)
if transactionIdentifier > math.MaxUint8 {
transactionIdentifier = 0
diff --git a/plc4go/internal/plc4go/s7/Connection.go b/plc4go/internal/plc4go/s7/Connection.go
index d2d0191..2c976c4 100644
--- a/plc4go/internal/plc4go/s7/Connection.go
+++ b/plc4go/internal/plc4go/s7/Connection.go
@@ -437,7 +437,7 @@ func (m Connection) ReadRequestBuilder() apiModel.PlcReadRequestBuilder {
func (m Connection) WriteRequestBuilder() apiModel.PlcWriteRequestBuilder {
return internalModel.NewDefaultPlcWriteRequestBuilder(
- m.fieldHandler, m.valueHandler, NewWriter(m.messageCodec))
+ m.fieldHandler, m.valueHandler, NewWriter(&m.tpduGenerator, m.messageCodec, m.tm))
}
func (m Connection) SubscriptionRequestBuilder() apiModel.PlcSubscriptionRequestBuilder {
diff --git a/plc4go/internal/plc4go/s7/Field.go b/plc4go/internal/plc4go/s7/Field.go
index e2618f6..601c5e5 100644
--- a/plc4go/internal/plc4go/s7/Field.go
+++ b/plc4go/internal/plc4go/s7/Field.go
@@ -27,6 +27,15 @@ import (
"github.com/rs/zerolog/log"
)
+type S7PlcField interface {
+ GetDataType() readWrite.TransportSize
+ GetNumElements() uint16
+ GetBlockNumber() uint16
+ GetMemoryArea() readWrite.MemoryArea
+ GetByteOffset() uint16
+ GetBitOffset() uint8
+}
+
type PlcField struct {
FieldType FieldType
MemoryArea readWrite.MemoryArea
@@ -82,6 +91,26 @@ func (m PlcField) GetDataType() readWrite.TransportSize {
return m.Datatype
}
+func (m PlcField) GetNumElements() uint16 {
+ return m.NumElements
+}
+
+func (m PlcField) GetBlockNumber() uint16 {
+ return m.BlockNumber
+}
+
+func (m PlcField) GetMemoryArea() readWrite.MemoryArea {
+ return m.MemoryArea
+}
+
+func (m PlcField) GetByteOffset() uint16 {
+ return m.ByteOffset
+}
+
+func (m PlcField) GetBitOffset() uint8 {
+ return m.BitOffset
+}
+
func (m PlcField) GetQuantity() uint16 {
return m.NumElements
}
diff --git a/plc4go/internal/plc4go/s7/Reader.go b/plc4go/internal/plc4go/s7/Reader.go
index e88ae00..2c6b73c 100644
--- a/plc4go/internal/plc4go/s7/Reader.go
+++ b/plc4go/internal/plc4go/s7/Reader.go
@@ -83,7 +83,11 @@ func (m *Reader) Read(readRequest model.PlcReadRequest) <-chan model.PlcReadRequ
log.Trace().Msg("Assemble paket")
// TODO: why do we use a uint16 above and the cotp a uint8?
tpktPacket := readWriteModel.NewTPKTPacket(
- readWriteModel.NewCOTPPacketData(true, uint8(tpduId), nil, s7MessageRequest),
+ readWriteModel.NewCOTPPacketData(true,
+ uint8(tpduId),
+ nil,
+ s7MessageRequest,
+ ),
)
// Start a new request-transaction (Is ended in the response-handler)
transaction := m.tm.StartRequest()
@@ -123,14 +127,12 @@ func (m *Reader) Read(readRequest model.PlcReadRequest) <-chan model.PlcReadRequ
Request: readRequest,
Err: errors.Wrap(err, "Error decoding response"),
}
- // TODO: should we return the error here?
- return nil
+ return transaction.EndRequest()
}
result <- model.PlcReadRequestResult{
Request: readRequest,
Response: readResponse,
}
- // TODO: check if we need to end transactions in error cases too
return transaction.EndRequest()
},
func(err error) error {
@@ -138,7 +140,7 @@ func (m *Reader) Read(readRequest model.PlcReadRequest) <-chan model.PlcReadRequ
Request: readRequest,
Err: errors.Wrap(err, "got timeout while waiting for response"),
}
- return nil
+ return transaction.EndRequest()
},
time.Second*1); err != nil {
result <- model.PlcReadRequestResult{
@@ -146,6 +148,7 @@ func (m *Reader) Read(readRequest model.PlcReadRequest) <-chan model.PlcReadRequ
Response: nil,
Err: errors.Wrap(err, "error sending message"),
}
+ _ = transaction.EndRequest()
}
})
}()
@@ -233,12 +236,12 @@ func (m *Reader) ToPlc4xReadResponse(response readWriteModel.S7Message, readRequ
// Currently we only support the S7 Any type of addresses. This helper simply converts the S7Field from PLC4X into
// S7Address objects.
func encodeS7Address(field model.PlcField) (*readWriteModel.S7Address, error) {
- s7Field, ok := field.(PlcField)
+ s7Field, ok := field.(S7PlcField)
if !ok {
return nil, errors.Errorf("Unsupported address type %t", field)
}
- transportSize := s7Field.Datatype
- numElements := s7Field.NumElements
+ transportSize := s7Field.GetDataType()
+ numElements := s7Field.GetNumElements()
// For these date-types we have to convert the requests to simple byte-array requests
// As otherwise the S7 will deny them with "Data type not supported" replies.
if (transportSize == readWriteModel.TransportSize_TIME) /*|| (transportSize == TransportSize.S7_S5TIME)*/ ||
@@ -265,10 +268,10 @@ func encodeS7Address(field model.PlcField) (*readWriteModel.S7Address, error) {
return readWriteModel.NewS7AddressAny(
transportSize,
numElements,
- s7Field.BlockNumber,
- s7Field.MemoryArea,
- s7Field.ByteOffset,
- s7Field.BitOffset,
+ s7Field.GetBlockNumber(),
+ s7Field.GetMemoryArea(),
+ s7Field.GetByteOffset(),
+ s7Field.GetBitOffset(),
), nil
}
diff --git a/plc4go/internal/plc4go/s7/Writer.go b/plc4go/internal/plc4go/s7/Writer.go
index 23b0280..9ac4531 100644
--- a/plc4go/internal/plc4go/s7/Writer.go
+++ b/plc4go/internal/plc4go/s7/Writer.go
@@ -24,190 +24,224 @@ import (
plc4goModel "github.com/apache/plc4x/plc4go/internal/plc4go/spi/model"
"github.com/apache/plc4x/plc4go/internal/plc4go/spi/utils"
"github.com/apache/plc4x/plc4go/pkg/plc4go/model"
+ "github.com/apache/plc4x/plc4go/pkg/plc4go/values"
"github.com/pkg/errors"
"github.com/rs/zerolog/log"
"time"
)
type Writer struct {
- messageCodec spi.MessageCodec
+ tpduGenerator *TpduGenerator
+ messageCodec spi.MessageCodec
+ tm *spi.RequestTransactionManager
}
-func NewWriter(messageCodec spi.MessageCodec) Writer {
+func NewWriter(tpduGenerator *TpduGenerator, messageCodec spi.MessageCodec, tm *spi.RequestTransactionManager) Writer {
return Writer{
- messageCodec: messageCodec,
+ tpduGenerator: tpduGenerator,
+ messageCodec: messageCodec,
+ tm: tm,
}
}
func (m Writer) Write(writeRequest model.PlcWriteRequest) <-chan model.PlcWriteRequestResult {
result := make(chan model.PlcWriteRequestResult)
go func() {
- // If we are requesting only one field, use a
- if len(writeRequest.GetFieldNames()) != 1 {
- result <- model.PlcWriteRequestResult{
- Request: writeRequest,
- Response: nil,
- Err: errors.New("s7 only supports single-item requests"),
- }
- return
- }
- fieldName := writeRequest.GetFieldNames()[0]
-
- // Get the s7 field instance from the request
- field := writeRequest.GetField(fieldName)
- s7Field, err := CastTos7FieldFromPlcField(field)
- if err != nil {
- result <- model.PlcWriteRequestResult{
- Request: writeRequest,
- Response: nil,
- Err: errors.Wrap(err, "invalid field item type"),
- }
- return
- }
- stringLength := uint16(0)
- if s7StringField, ok := field.(*PlcStringField); ok {
- stringLength = s7StringField.stringLength
- }
-
- // Get the value from the request and serialize it to a byte array
- value := writeRequest.GetValue(fieldName)
- io := utils.NewWriteBuffer()
- if err := readWriteModel.DataItemSerialize(io, value, s7Field.Datatype.DataProtocolId(), int32(stringLength)); err != nil {
- result <- model.PlcWriteRequestResult{
- Request: writeRequest,
- Response: nil,
- Err: errors.Wrap(err, "error serializing value"),
- }
- return
- }
- utils.Uint8ArrayToInt8Array(io.GetBytes())
-
- // TODO disect child type
- /*
- var pdu *readWriteModel.s7PDU
- switch s7Field.FieldType {
- case Coil:
- pdu = readWriteModel.News7PDUWriteMultipleCoilsRequest(
- s7Field.Address,
- s7Field.Quantity,
- data)
- case HoldingRegister:
- pdu = readWriteModel.News7PDUWriteMultipleHoldingRegistersRequest(
- s7Field.Address,
- numWords,
- data)
- case ExtendedRegister:
+ parameterItems := make([]*readWriteModel.S7VarRequestParameterItem, len(writeRequest.GetFieldNames()))
+ payloadItems := make([]*readWriteModel.S7VarPayloadDataItem, len(writeRequest.GetFieldNames()))
+ for i, fieldName := range writeRequest.GetFieldNames() {
+ field := writeRequest.GetField(fieldName)
+ plcValue := writeRequest.GetValue(fieldName)
+ s7Address, err := encodeS7Address(field)
+ if err != nil {
result <- model.PlcWriteRequestResult{
Request: writeRequest,
Response: nil,
- Err: errors.New("s7 currently doesn't support extended register requests"),
+ Err: errors.Wrapf(err, "Error encoding s7 address for field %s", fieldName),
}
return
- default:
+ }
+ parameterItems[i] = readWriteModel.NewS7VarRequestParameterItemAddress(s7Address)
+ value, err := serializePlcValue(field, plcValue)
+ if err != nil {
result <- model.PlcWriteRequestResult{
Request: writeRequest,
Response: nil,
- Err: errors.New("unsupported field type"),
+ Err: errors.Wrapf(err, "Error encoding value for field %s", fieldName),
}
return
}
- */
-
- // Assemble the paket
- request := readWriteModel.S7MessageRequest{}
-
- // Send the over the wire
- err = m.messageCodec.SendRequest(
- request,
- func(message interface{}) bool {
- readWriteModel.CastS7MessageResponse(message)
- // TODO: match to message
- return false
- },
- func(message interface{}) error {
- // Convert the response into an
- response := readWriteModel.CastS7MessageResponse(message)
- // Convert the s7 response into a PLC4X response
- readResponse, err := m.ToPlc4xWriteResponse(request, *response, writeRequest)
-
- if err != nil {
- result <- model.PlcWriteRequestResult{
- Request: writeRequest,
- Err: errors.Wrap(err, "Error decoding response"),
+ payloadItems[i] = value
+ }
+ tpduId := m.tpduGenerator.getAndIncrement()
+
+ // Create a new Request with correct tpuId (is not known before)
+ s7MessageRequest := readWriteModel.NewS7MessageRequest(
+ tpduId,
+ readWriteModel.NewS7ParameterWriteVarRequest(parameterItems),
+ readWriteModel.NewS7PayloadWriteVarRequest(payloadItems),
+ )
+
+ // Assemble the finished paket
+ log.Trace().Msg("Assemble paket")
+ // TODO: why do we use a uint16 above and the cotp a uint8?
+ tpktPacket := readWriteModel.NewTPKTPacket(
+ readWriteModel.NewCOTPPacketData(
+ true,
+ uint8(tpduId),
+ nil,
+ s7MessageRequest,
+ ),
+ )
+
+ // Start a new request-transaction (Is ended in the response-handler)
+ transaction := m.tm.StartRequest()
+ transaction.Submit(func() {
+ // Send the over the wire
+ if err := m.messageCodec.SendRequest(
+ tpktPacket,
+ func(message interface{}) bool {
+ tpktPacket := readWriteModel.CastTPKTPacket(message)
+ if tpktPacket == nil {
+ return false
+ }
+ cotpPacketData := readWriteModel.CastCOTPPacketData(tpktPacket.Payload)
+ if cotpPacketData == nil {
+ return false
+ }
+ payload := cotpPacketData.Parent.Payload
+ if payload == nil {
+ return false
+ }
+ return payload.TpduReference == tpduId
+ },
+ func(message interface{}) error {
+ // Convert the response into an
+ log.Trace().Msg("convert response to ")
+ tpktPacket := readWriteModel.CastTPKTPacket(message)
+ cotpPacketData := readWriteModel.CastCOTPPacketData(tpktPacket.Payload)
+ payload := cotpPacketData.Parent.Payload
+ // Convert the s7 response into a PLC4X response
+ log.Trace().Msg("convert response to PLC4X response")
+ readResponse, err := m.ToPlc4xWriteResponse(*payload, writeRequest)
+
+ if err != nil {
+ result <- model.PlcWriteRequestResult{
+ Request: writeRequest,
+ Err: errors.Wrap(err, "Error decoding response"),
+ }
+ return transaction.EndRequest()
}
- } else {
result <- model.PlcWriteRequestResult{
Request: writeRequest,
Response: readResponse,
}
- }
- return nil
- },
- func(err error) error {
+ return transaction.EndRequest()
+ },
+ func(err error) error {
+ result <- model.PlcWriteRequestResult{
+ Request: writeRequest,
+ Err: errors.New("got timeout while waiting for response"),
+ }
+ return transaction.EndRequest()
+ },
+ time.Second*1); err != nil {
result <- model.PlcWriteRequestResult{
- Request: writeRequest,
- Err: errors.New("got timeout while waiting for response"),
+ Request: writeRequest,
+ Response: nil,
+ Err: errors.Wrap(err, "error sending message"),
}
- return nil
- },
- time.Second*1)
+ _ = transaction.EndRequest()
+ }
+ })
}()
return result
}
-func (m Writer) ToPlc4xWriteResponse(request readWriteModel.S7MessageRequest, response readWriteModel.S7MessageResponse, writeRequest model.PlcWriteRequest) (model.PlcWriteResponse, error) {
+func (m Writer) ToPlc4xWriteResponse(response readWriteModel.S7Message, writeRequest model.PlcWriteRequest) (model.PlcWriteResponse, error) {
+ var errorClass uint8
+ var errorCode uint8
+ switch response.Child.(type) {
+ case *readWriteModel.S7MessageResponseData:
+ messageResponseData := response.Child.(*readWriteModel.S7MessageResponseData)
+ errorClass = messageResponseData.ErrorClass
+ errorCode = messageResponseData.ErrorCode
+ case *readWriteModel.S7MessageResponse:
+ messageResponseData := response.Child.(*readWriteModel.S7MessageResponse)
+ errorClass = messageResponseData.ErrorClass
+ errorCode = messageResponseData.ErrorCode
+ default:
+ return nil, errors.Errorf("unsupported response type %T", response.Child)
+ }
responseCodes := map[string]model.PlcResponseCode{}
- fieldName := writeRequest.GetFieldNames()[0]
-
- // we default to an error until its proven wrong
- responseCodes[fieldName] = model.PlcResponseCode_INTERNAL_ERROR
- // TODO disect child type
- /*
- switch response.Pdu.Child.(type) {
- case *readWriteModel.s7PDUWriteMultipleCoilsResponse:
- req := readWriteModel.Casts7PDUWriteMultipleCoilsRequest(request.Pdu)
- resp := readWriteModel.Casts7PDUWriteMultipleCoilsResponse(response.Pdu)
- if req.Quantity == resp.Quantity {
- responseCodes[fieldName] = model.PlcResponseCode_OK
- }
- case *readWriteModel.s7PDUWriteMultipleHoldingRegistersResponse:
- req := readWriteModel.Casts7PDUWriteMultipleHoldingRegistersRequest(request.Pdu)
- resp := readWriteModel.Casts7PDUWriteMultipleHoldingRegistersResponse(response.Pdu)
- if req.Quantity == resp.Quantity {
- responseCodes[fieldName] = model.PlcResponseCode_OK
+
+ // If the result contains any form of non-null error code, handle this instead.
+ if (errorClass != 0) || (errorCode != 0) {
+ // This is usually the case if PUT/GET wasn't enabled on the PLC
+ if (errorClass == 129) && (errorCode == 4) {
+ log.Warn().Msg("Got an error response from the PLC. This particular response code usually indicates " +
+ "that PUT/GET is not enabled on the PLC.")
+ for _, fieldName := range writeRequest.GetFieldNames() {
+ responseCodes[fieldName] = model.PlcResponseCode_ACCESS_DENIED
}
- case *readWriteModel.s7PDUError:
- resp := readWriteModel.Casts7PDUError(&response.Pdu)
- switch resp.ExceptionCode {
- case readWriteModel.s7ErrorCode_ILLEGAL_FUNCTION:
- responseCodes[fieldName] = model.PlcResponseCode_UNSUPPORTED
- case readWriteModel.s7ErrorCode_ILLEGAL_DATA_ADDRESS:
- responseCodes[fieldName] = model.PlcResponseCode_INVALID_ADDRESS
- case readWriteModel.s7ErrorCode_ILLEGAL_DATA_VALUE:
- responseCodes[fieldName] = model.PlcResponseCode_INVALID_DATA
- case readWriteModel.s7ErrorCode_SLAVE_DEVICE_FAILURE:
- responseCodes[fieldName] = model.PlcResponseCode_REMOTE_ERROR
- case readWriteModel.s7ErrorCode_ACKNOWLEDGE:
- responseCodes[fieldName] = model.PlcResponseCode_OK
- case readWriteModel.s7ErrorCode_SLAVE_DEVICE_BUSY:
- responseCodes[fieldName] = model.PlcResponseCode_REMOTE_BUSY
- case readWriteModel.s7ErrorCode_NEGATIVE_ACKNOWLEDGE:
- responseCodes[fieldName] = model.PlcResponseCode_REMOTE_ERROR
- case readWriteModel.s7ErrorCode_MEMORY_PARITY_ERROR:
- responseCodes[fieldName] = model.PlcResponseCode_INTERNAL_ERROR
- case readWriteModel.s7ErrorCode_GATEWAY_PATH_UNAVAILABLE:
+ log.Trace().Msg("Returning the response")
+ return plc4goModel.NewDefaultPlcWriteResponse(writeRequest, responseCodes), nil
+ } else {
+ log.Warn().Msgf("Got an unknown error response from the PLC. Error Class: %d, Error Code %d. "+
+ "We probably need to implement explicit handling for this, so please file a bug-report "+
+ "on https://issues.apache.org/jira/projects/PLC4X and ideally attach a WireShark dump "+
+ "containing a capture of the communication.",
+ errorClass, errorCode)
+ for _, fieldName := range writeRequest.GetFieldNames() {
responseCodes[fieldName] = model.PlcResponseCode_INTERNAL_ERROR
- case readWriteModel.s7ErrorCode_GATEWAY_TARGET_DEVICE_FAILED_TO_RESPOND:
- responseCodes[fieldName] = model.PlcResponseCode_REMOTE_ERROR
- default:
- log.Debug().Msgf("Unmapped exception code %x", resp.ExceptionCode)
}
- default:
- return nil, errors.Errorf("unsupported response type %T", response.Pdu.Child)
+ return plc4goModel.NewDefaultPlcWriteResponse(writeRequest, responseCodes), nil
}
- */
+ }
+
+ // In all other cases all went well.
+ payload := response.Payload.Child.(*readWriteModel.S7PayloadWriteVarResponse)
+
+ // If the numbers of items don't match, we're in big trouble as the only
+ // way to know how to interpret the responses is by aligning them with the
+ // items from the request as this information is not returned by the PLC.
+ if len(writeRequest.GetFieldNames()) != len(payload.Items) {
+ return nil, errors.New("The number of requested items doesn't match the number of returned items")
+ }
+
+ payloadItems := payload.Items
+ for i, fieldName := range writeRequest.GetFieldNames() {
+ payloadItem := payloadItems[i]
+
+ responseCode := decodeResponseCode(payloadItem.ReturnCode)
+ // Decode the data according to the information from the request
+ log.Trace().Msg("decode data")
+ responseCodes[fieldName] = responseCode
+ }
// Return the response
log.Trace().Msg("Returning the response")
return plc4goModel.NewDefaultPlcWriteResponse(writeRequest, responseCodes), nil
}
+
+func serializePlcValue(field model.PlcField, plcValue values.PlcValue) (*readWriteModel.S7VarPayloadDataItem, error) {
+ s7Field, ok := field.(S7PlcField)
+ if !ok {
+ return nil, errors.Errorf("Unsupported address type %t", field)
+ }
+ transportSize := s7Field.GetDataType().DataTransportSize()
+ stringLength := uint16(254)
+ if s7StringField, ok := field.(*PlcStringField); ok {
+ stringLength = s7StringField.stringLength
+ }
+ io := utils.NewWriteBuffer()
+ err := readWriteModel.DataItemSerialize(io, plcValue, s7Field.GetDataType().DataProtocolId(), int32(stringLength))
+ if err != nil {
+ return nil, errors.Wrapf(err, "Error serializing field item of type: '%v'", s7Field.GetDataType())
+ }
+ data := io.GetBytes()
+ return readWriteModel.NewS7VarPayloadDataItem(
+ readWriteModel.DataTransportErrorCode_OK,
+ transportSize, utils.Uint8ArrayToInt8Array(data),
+ ), nil
+}