Code Duplication - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Code Duplication Length = 44-46 lines in 8 locations

opcua/ua/uaprotocol_auto.py 8 locations


        self.RequestHeader = RequestHeader()
        self.Parameters = RegisterNodesParameters()
        self._freeze = True

    def to_binary(self):
        packet = []
        packet.append(self.TypeId.to_binary())
        packet.append(self.RequestHeader.to_binary())
        packet.append(self.Parameters.to_binary())
        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        obj = RegisterNodesRequest()
        obj.TypeId = NodeId.from_binary(data)
        obj.RequestHeader = RequestHeader.from_binary(data)
        obj.Parameters = RegisterNodesParameters.from_binary(data)
        return obj

    def __str__(self):
        return 'RegisterNodesRequest(' + 'TypeId:' + str(self.TypeId) + ', ' + \
               'RequestHeader:' + str(self.RequestHeader) + ', ' + \
               'Parameters:' + str(self.Parameters) + ')'

    __repr__ = __str__


class RegisterNodesResult(FrozenClass):
    '''
    :ivar RegisteredNodeIds:
    :vartype RegisteredNodeIds: NodeId
    '''

    ua_types = [
        ('RegisteredNodeIds', 'ListOfNodeId'),
               ]

    def __init__(self):
        self.RegisteredNodeIds = []
        self._freeze = True

    def to_binary(self):
        packet = []
        packet.append(uabin.Primitives.Int32.pack(len(self.RegisteredNodeIds)))
        for fieldname in self.RegisteredNodeIds:
            packet.append(fieldname.to_binary())
        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        obj = RegisterNodesResult()
        length = uabin.Primitives.Int32.unpack(data)

        ('TranslateBrowsePathsToNodeIdsCount', 'ServiceCounterDataType'),
        ('QueryFirstCount', 'ServiceCounterDataType'),
        ('QueryNextCount', 'ServiceCounterDataType'),
        ('RegisterNodesCount', 'ServiceCounterDataType'),
        ('UnregisterNodesCount', 'ServiceCounterDataType'),
               ]

    def __init__(self):
        self.SessionId = NodeId()
        self.SessionName = None
        self.ClientDescription = ApplicationDescription()
        self.ServerUri = None
        self.EndpointUrl = None
        self.LocaleIds = []
        self.ActualSessionTimeout = 0
        self.MaxResponseMessageSize = 0
        self.ClientConnectionTime = datetime.utcnow()
        self.ClientLastContactTime = datetime.utcnow()
        self.CurrentSubscriptionsCount = 0
        self.CurrentMonitoredItemsCount = 0
        self.CurrentPublishRequestsInQueue = 0
        self.TotalRequestCount = ServiceCounterDataType()
        self.UnauthorizedRequestCount = 0
        self.ReadCount = ServiceCounterDataType()
        self.HistoryReadCount = ServiceCounterDataType()
        self.WriteCount = ServiceCounterDataType()
        self.HistoryUpdateCount = ServiceCounterDataType()
        self.CallCount = ServiceCounterDataType()
        self.CreateMonitoredItemsCount = ServiceCounterDataType()
        self.ModifyMonitoredItemsCount = ServiceCounterDataType()
        self.SetMonitoringModeCount = ServiceCounterDataType()
        self.SetTriggeringCount = ServiceCounterDataType()
        self.DeleteMonitoredItemsCount = ServiceCounterDataType()
        self.CreateSubscriptionCount = ServiceCounterDataType()
        self.ModifySubscriptionCount = ServiceCounterDataType()
        self.SetPublishingModeCount = ServiceCounterDataType()
        self.PublishCount = ServiceCounterDataType()
        self.RepublishCount = ServiceCounterDataType()
        self.TransferSubscriptionsCount = ServiceCounterDataType()
        self.DeleteSubscriptionsCount = ServiceCounterDataType()
        self.AddNodesCount = ServiceCounterDataType()
        self.AddReferencesCount = ServiceCounterDataType()
        self.DeleteNodesCount = ServiceCounterDataType()
        self.DeleteReferencesCount = ServiceCounterDataType()
        self.BrowseCount = ServiceCounterDataType()
        self.BrowseNextCount = ServiceCounterDataType()
        self.TranslateBrowsePathsToNodeIdsCount = ServiceCounterDataType()
        self.QueryFirstCount = ServiceCounterDataType()
        self.QueryNextCount = ServiceCounterDataType()
        self.RegisterNodesCount = ServiceCounterDataType()

        ('SubscriptionId', 'UInt32'),
        ('RetransmitSequenceNumber', 'UInt32'),
               ]

    def __init__(self):
        self.SubscriptionId = 0
        self.RetransmitSequenceNumber = 0
        self._freeze = True

    def to_binary(self):
        packet = []
        packet.append(uabin.Primitives.UInt32.pack(self.SubscriptionId))
        packet.append(uabin.Primitives.UInt32.pack(self.RetransmitSequenceNumber))
        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        obj = RepublishParameters()
        self.SubscriptionId = uabin.Primitives.UInt32.unpack(data)
        self.RetransmitSequenceNumber = uabin.Primitives.UInt32.unpack(data)
        return obj

    def __str__(self):
        return 'RepublishParameters(' + 'SubscriptionId:' + str(self.SubscriptionId) + ', ' + \
               'RetransmitSequenceNumber:' + str(self.RetransmitSequenceNumber) + ')'

    __repr__ = __str__


class RepublishRequest(FrozenClass):
    '''
    :ivar TypeId:
    :vartype TypeId: NodeId
    :ivar RequestHeader:
    :vartype RequestHeader: RequestHeader
    :ivar Parameters:
    :vartype Parameters: RepublishParameters
    '''

    ua_types = [
        ('TypeId', 'NodeId'),
        ('RequestHeader', 'RequestHeader'),
        ('Parameters', 'RepublishParameters'),
               ]

    def __init__(self):
        self.TypeId = FourByteNodeId(ObjectIds.RepublishRequest_Encoding_DefaultBinary)
        self.RequestHeader = RequestHeader()
        self.Parameters = RepublishParameters()
        self._freeze = True


    def __str__(self):
        return 'MonitoringParameters(' + 'ClientHandle:' + str(self.ClientHandle) + ', ' + \
               'SamplingInterval:' + str(self.SamplingInterval) + ', ' + \
               'Filter:' + str(self.Filter) + ', ' + \
               'QueueSize:' + str(self.QueueSize) + ', ' + \
               'DiscardOldest:' + str(self.DiscardOldest) + ')'

    __repr__ = __str__


class MonitoredItemCreateRequest(FrozenClass):
    '''
    :ivar ItemToMonitor:
    :vartype ItemToMonitor: ReadValueId
    :ivar MonitoringMode:
    :vartype MonitoringMode: MonitoringMode
    :ivar RequestedParameters:
    :vartype RequestedParameters: MonitoringParameters
    '''

    ua_types = [
        ('ItemToMonitor', 'ReadValueId'),
        ('MonitoringMode', 'MonitoringMode'),
        ('RequestedParameters', 'MonitoringParameters'),
               ]

    def __init__(self):
        self.ItemToMonitor = ReadValueId()
        self.MonitoringMode = MonitoringMode(0)
        self.RequestedParameters = MonitoringParameters()
        self._freeze = True

    def to_binary(self):
        packet = []
        packet.append(self.ItemToMonitor.to_binary())
        packet.append(uabin.Primitives.UInt32.pack(self.MonitoringMode.value))
        packet.append(self.RequestedParameters.to_binary())
        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        obj = MonitoredItemCreateRequest()
        obj.ItemToMonitor = ReadValueId.from_binary(data)
        self.MonitoringMode = MonitoringMode(uabin.Primitives.UInt32.unpack(data))
        obj.RequestedParameters = MonitoringParameters.from_binary(data)
        return obj

    def __str__(self):
        return 'MonitoredItemCreateRequest(' + 'ItemToMonitor:' + str(self.ItemToMonitor) + ', ' + \

            for _ in range(0, length):
                array.append(DiagnosticInfo.from_binary(data))
        obj.DiagnosticInfos = array
        return obj

    def __str__(self):
        return 'ReadResponse(' + 'TypeId:' + str(self.TypeId) + ', ' + \
               'ResponseHeader:' + str(self.ResponseHeader) + ', ' + \
               'Results:' + str(self.Results) + ', ' + \
               'DiagnosticInfos:' + str(self.DiagnosticInfos) + ')'

    __repr__ = __str__


class HistoryReadValueId(FrozenClass):
    '''
    :ivar NodeId:
    :vartype NodeId: NodeId
    :ivar IndexRange:
    :vartype IndexRange: String
    :ivar DataEncoding:
    :vartype DataEncoding: QualifiedName
    :ivar ContinuationPoint:
    :vartype ContinuationPoint: ByteString
    '''

    ua_types = [
        ('NodeId', 'NodeId'),
        ('IndexRange', 'String'),
        ('DataEncoding', 'QualifiedName'),
        ('ContinuationPoint', 'ByteString'),
               ]

    def __init__(self):
        self.NodeId = NodeId()
        self.IndexRange = None
        self.DataEncoding = QualifiedName()
        self.ContinuationPoint = None
        self._freeze = True

    def to_binary(self):
        packet = []
        packet.append(self.NodeId.to_binary())
        packet.append(uabin.Primitives.String.pack(self.IndexRange))
        packet.append(self.DataEncoding.to_binary())
        packet.append(uabin.Primitives.ByteString.pack(self.ContinuationPoint))
        return b''.join(packet)

    @staticmethod
    def from_binary(data):


    ua_types = [
        ('EndpointUrl', 'String'),
        ('Server', 'ApplicationDescription'),
        ('ServerCertificate', 'ByteString'),
        ('SecurityMode', 'MessageSecurityMode'),
        ('SecurityPolicyUri', 'String'),
        ('UserIdentityTokens', 'ListOfUserTokenPolicy'),
        ('TransportProfileUri', 'String'),
        ('SecurityLevel', 'Byte'),
               ]

    def __init__(self):
        self.EndpointUrl = None
        self.Server = ApplicationDescription()
        self.ServerCertificate = None
        self.SecurityMode = MessageSecurityMode(0)
        self.SecurityPolicyUri = None
        self.UserIdentityTokens = []
        self.TransportProfileUri = None
        self.SecurityLevel = 0
        self._freeze = True

    def to_binary(self):
        packet = []
        packet.append(uabin.Primitives.String.pack(self.EndpointUrl))
        packet.append(self.Server.to_binary())
        packet.append(uabin.Primitives.ByteString.pack(self.ServerCertificate))
        packet.append(uabin.Primitives.UInt32.pack(self.SecurityMode.value))
        packet.append(uabin.Primitives.String.pack(self.SecurityPolicyUri))
        packet.append(uabin.Primitives.Int32.pack(len(self.UserIdentityTokens)))
        for fieldname in self.UserIdentityTokens:
            packet.append(fieldname.to_binary())
        packet.append(uabin.Primitives.String.pack(self.TransportProfileUri))
        packet.append(uabin.Primitives.Byte.pack(self.SecurityLevel))
        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        obj = EndpointDescription()
        self.EndpointUrl = uabin.Primitives.String.unpack(data)
        obj.Server = ApplicationDescription.from_binary(data)
        self.ServerCertificate = uabin.Primitives.ByteString.unpack(data)
        self.SecurityMode = MessageSecurityMode(uabin.Primitives.UInt32.unpack(data))
        self.SecurityPolicyUri = uabin.Primitives.String.unpack(data)
        length = uabin.Primitives.Int32.unpack(data)
        array = []
        if length != -1:
            for _ in range(0, length):
                array.append(UserTokenPolicy.from_binary(data))

        ('OperandDiagnosticInfos', 'ListOfDiagnosticInfo'),
               ]

    def __init__(self):
        self.StatusCode = StatusCode()
        self.OperandStatusCodes = []
        self.OperandDiagnosticInfos = []
        self._freeze = True

    def to_binary(self):
        packet = []
        packet.append(self.StatusCode.to_binary())
        packet.append(uabin.Primitives.Int32.pack(len(self.OperandStatusCodes)))
        for fieldname in self.OperandStatusCodes:
            packet.append(fieldname.to_binary())
        packet.append(uabin.Primitives.Int32.pack(len(self.OperandDiagnosticInfos)))
        for fieldname in self.OperandDiagnosticInfos:
            packet.append(fieldname.to_binary())
        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        obj = ContentFilterElementResult()
        obj.StatusCode = StatusCode.from_binary(data)
        length = uabin.Primitives.Int32.unpack(data)
        array = []
        if length != -1:
            for _ in range(0, length):
                array.append(StatusCode.from_binary(data))
        obj.OperandStatusCodes = array
        length = uabin.Primitives.Int32.unpack(data)
        array = []
        if length != -1:
            for _ in range(0, length):
                array.append(DiagnosticInfo.from_binary(data))
        obj.OperandDiagnosticInfos = array
        return obj

    def __str__(self):
        return 'ContentFilterElementResult(' + 'StatusCode:' + str(self.StatusCode) + ', ' + \
               'OperandStatusCodes:' + str(self.OperandStatusCodes) + ', ' + \
               'OperandDiagnosticInfos:' + str(self.OperandDiagnosticInfos) + ')'

    __repr__ = __str__


class ContentFilterResult(FrozenClass):
    '''
    :ivar ElementResults:
    :vartype ElementResults: ContentFilterElementResult

    def to_binary(self):
        packet = []
        packet.append(uabin.Primitives.UInt32.pack(self.ChannelId))
        packet.append(uabin.Primitives.UInt32.pack(self.TokenId))
        packet.append(uabin.Primitives.DateTime.pack(self.CreatedAt))
        packet.append(uabin.Primitives.UInt32.pack(self.RevisedLifetime))
        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        obj = ChannelSecurityToken()
        self.ChannelId = uabin.Primitives.UInt32.unpack(data)
        self.TokenId = uabin.Primitives.UInt32.unpack(data)
        self.CreatedAt = uabin.Primitives.DateTime.unpack(data)
        self.RevisedLifetime = uabin.Primitives.UInt32.unpack(data)
        return obj

    def __str__(self):
        return 'ChannelSecurityToken(' + 'ChannelId:' + str(self.ChannelId) + ', ' + \
               'TokenId:' + str(self.TokenId) + ', ' + \
               'CreatedAt:' + str(self.CreatedAt) + ', ' + \
               'RevisedLifetime:' + str(self.RevisedLifetime) + ')'

    __repr__ = __str__


class OpenSecureChannelParameters(FrozenClass):
    '''
    :ivar ClientProtocolVersion:
    :vartype ClientProtocolVersion: UInt32
    :ivar RequestType:
    :vartype RequestType: SecurityTokenRequestType
    :ivar SecurityMode:
    :vartype SecurityMode: MessageSecurityMode
    :ivar ClientNonce:
    :vartype ClientNonce: ByteString
    :ivar RequestedLifetime:
    :vartype RequestedLifetime: UInt32
    '''

    ua_types = [
        ('ClientProtocolVersion', 'UInt32'),
        ('RequestType', 'SecurityTokenRequestType'),
        ('SecurityMode', 'MessageSecurityMode'),
        ('ClientNonce', 'ByteString'),
        ('RequestedLifetime', 'UInt32'),
               ]

    def __init__(self):
        self.ClientProtocolVersion = 0

		@@ 6605-6650 (lines=46) @@
6602		self.RequestHeader = RequestHeader()
6603		self.Parameters = RegisterNodesParameters()
6604		self._freeze = True
6605
6606		def to_binary(self):
6607		packet = []
6608		packet.append(self.TypeId.to_binary())
6609		packet.append(self.RequestHeader.to_binary())
6610		packet.append(self.Parameters.to_binary())
6611		return b''.join(packet)
6612
6613		@staticmethod
6614		def from_binary(data):
6615		obj = RegisterNodesRequest()
6616		obj.TypeId = NodeId.from_binary(data)
6617		obj.RequestHeader = RequestHeader.from_binary(data)
6618		obj.Parameters = RegisterNodesParameters.from_binary(data)
6619		return obj
6620
6621		def __str__(self):
6622		return 'RegisterNodesRequest(' + 'TypeId:' + str(self.TypeId) + ', ' + \
6623		'RequestHeader:' + str(self.RequestHeader) + ', ' + \
6624		'Parameters:' + str(self.Parameters) + ')'
6625
6626		__repr__ = __str__
6627
6628
6629		class RegisterNodesResult(FrozenClass):
6630		'''
6631		:ivar RegisteredNodeIds:
6632		:vartype RegisteredNodeIds: NodeId
6633		'''
6634
6635		ua_types = [
6636		('RegisteredNodeIds', 'ListOfNodeId'),
6637		]
6638
6639		def __init__(self):
6640		self.RegisteredNodeIds = []
6641		self._freeze = True
6642
6643		def to_binary(self):
6644		packet = []
6645		packet.append(uabin.Primitives.Int32.pack(len(self.RegisteredNodeIds)))
6646		for fieldname in self.RegisteredNodeIds:
6647		packet.append(fieldname.to_binary())
6648		return b''.join(packet)
6649
6650		@staticmethod
6651		def from_binary(data):
6652		obj = RegisterNodesResult()
6653		length = uabin.Primitives.Int32.unpack(data)
		@@ 13757-13800 (lines=44) @@
13754		('TranslateBrowsePathsToNodeIdsCount', 'ServiceCounterDataType'),
13755		('QueryFirstCount', 'ServiceCounterDataType'),
13756		('QueryNextCount', 'ServiceCounterDataType'),
13757		('RegisterNodesCount', 'ServiceCounterDataType'),
13758		('UnregisterNodesCount', 'ServiceCounterDataType'),
13759		]
13760
13761		def __init__(self):
13762		self.SessionId = NodeId()
13763		self.SessionName = None
13764		self.ClientDescription = ApplicationDescription()
13765		self.ServerUri = None
13766		self.EndpointUrl = None
13767		self.LocaleIds = []
13768		self.ActualSessionTimeout = 0
13769		self.MaxResponseMessageSize = 0
13770		self.ClientConnectionTime = datetime.utcnow()
13771		self.ClientLastContactTime = datetime.utcnow()
13772		self.CurrentSubscriptionsCount = 0
13773		self.CurrentMonitoredItemsCount = 0
13774		self.CurrentPublishRequestsInQueue = 0
13775		self.TotalRequestCount = ServiceCounterDataType()
13776		self.UnauthorizedRequestCount = 0
13777		self.ReadCount = ServiceCounterDataType()
13778		self.HistoryReadCount = ServiceCounterDataType()
13779		self.WriteCount = ServiceCounterDataType()
13780		self.HistoryUpdateCount = ServiceCounterDataType()
13781		self.CallCount = ServiceCounterDataType()
13782		self.CreateMonitoredItemsCount = ServiceCounterDataType()
13783		self.ModifyMonitoredItemsCount = ServiceCounterDataType()
13784		self.SetMonitoringModeCount = ServiceCounterDataType()
13785		self.SetTriggeringCount = ServiceCounterDataType()
13786		self.DeleteMonitoredItemsCount = ServiceCounterDataType()
13787		self.CreateSubscriptionCount = ServiceCounterDataType()
13788		self.ModifySubscriptionCount = ServiceCounterDataType()
13789		self.SetPublishingModeCount = ServiceCounterDataType()
13790		self.PublishCount = ServiceCounterDataType()
13791		self.RepublishCount = ServiceCounterDataType()
13792		self.TransferSubscriptionsCount = ServiceCounterDataType()
13793		self.DeleteSubscriptionsCount = ServiceCounterDataType()
13794		self.AddNodesCount = ServiceCounterDataType()
13795		self.AddReferencesCount = ServiceCounterDataType()
13796		self.DeleteNodesCount = ServiceCounterDataType()
13797		self.DeleteReferencesCount = ServiceCounterDataType()
13798		self.BrowseCount = ServiceCounterDataType()
13799		self.BrowseNextCount = ServiceCounterDataType()
13800		self.TranslateBrowsePathsToNodeIdsCount = ServiceCounterDataType()
13801		self.QueryFirstCount = ServiceCounterDataType()
13802		self.QueryNextCount = ServiceCounterDataType()
13803		self.RegisterNodesCount = ServiceCounterDataType()
		@@ 12728-12771 (lines=44) @@
12725		('SubscriptionId', 'UInt32'),
12726		('RetransmitSequenceNumber', 'UInt32'),
12727		]
12728
12729		def __init__(self):
12730		self.SubscriptionId = 0
12731		self.RetransmitSequenceNumber = 0
12732		self._freeze = True
12733
12734		def to_binary(self):
12735		packet = []
12736		packet.append(uabin.Primitives.UInt32.pack(self.SubscriptionId))
12737		packet.append(uabin.Primitives.UInt32.pack(self.RetransmitSequenceNumber))
12738		return b''.join(packet)
12739
12740		@staticmethod
12741		def from_binary(data):
12742		obj = RepublishParameters()
12743		self.SubscriptionId = uabin.Primitives.UInt32.unpack(data)
12744		self.RetransmitSequenceNumber = uabin.Primitives.UInt32.unpack(data)
12745		return obj
12746
12747		def __str__(self):
12748		return 'RepublishParameters(' + 'SubscriptionId:' + str(self.SubscriptionId) + ', ' + \
12749		'RetransmitSequenceNumber:' + str(self.RetransmitSequenceNumber) + ')'
12750
12751		__repr__ = __str__
12752
12753
12754		class RepublishRequest(FrozenClass):
12755		'''
12756		:ivar TypeId:
12757		:vartype TypeId: NodeId
12758		:ivar RequestHeader:
12759		:vartype RequestHeader: RequestHeader
12760		:ivar Parameters:
12761		:vartype Parameters: RepublishParameters
12762		'''
12763
12764		ua_types = [
12765		('TypeId', 'NodeId'),
12766		('RequestHeader', 'RequestHeader'),
12767		('Parameters', 'RepublishParameters'),
12768		]
12769
12770		def __init__(self):
12771		self.TypeId = FourByteNodeId(ObjectIds.RepublishRequest_Encoding_DefaultBinary)
12772		self.RequestHeader = RequestHeader()
12773		self.Parameters = RepublishParameters()
12774		self._freeze = True
		@@ 10457-10500 (lines=44) @@
10454
10455		def __str__(self):
10456		return 'MonitoringParameters(' + 'ClientHandle:' + str(self.ClientHandle) + ', ' + \
10457		'SamplingInterval:' + str(self.SamplingInterval) + ', ' + \
10458		'Filter:' + str(self.Filter) + ', ' + \
10459		'QueueSize:' + str(self.QueueSize) + ', ' + \
10460		'DiscardOldest:' + str(self.DiscardOldest) + ')'
10461
10462		__repr__ = __str__
10463
10464
10465		class MonitoredItemCreateRequest(FrozenClass):
10466		'''
10467		:ivar ItemToMonitor:
10468		:vartype ItemToMonitor: ReadValueId
10469		:ivar MonitoringMode:
10470		:vartype MonitoringMode: MonitoringMode
10471		:ivar RequestedParameters:
10472		:vartype RequestedParameters: MonitoringParameters
10473		'''
10474
10475		ua_types = [
10476		('ItemToMonitor', 'ReadValueId'),
10477		('MonitoringMode', 'MonitoringMode'),
10478		('RequestedParameters', 'MonitoringParameters'),
10479		]
10480
10481		def __init__(self):
10482		self.ItemToMonitor = ReadValueId()
10483		self.MonitoringMode = MonitoringMode(0)
10484		self.RequestedParameters = MonitoringParameters()
10485		self._freeze = True
10486
10487		def to_binary(self):
10488		packet = []
10489		packet.append(self.ItemToMonitor.to_binary())
10490		packet.append(uabin.Primitives.UInt32.pack(self.MonitoringMode.value))
10491		packet.append(self.RequestedParameters.to_binary())
10492		return b''.join(packet)
10493
10494		@staticmethod
10495		def from_binary(data):
10496		obj = MonitoredItemCreateRequest()
10497		obj.ItemToMonitor = ReadValueId.from_binary(data)
10498		self.MonitoringMode = MonitoringMode(uabin.Primitives.UInt32.unpack(data))
10499		obj.RequestedParameters = MonitoringParameters.from_binary(data)
10500		return obj
10501
10502		def __str__(self):
10503		return 'MonitoredItemCreateRequest(' + 'ItemToMonitor:' + str(self.ItemToMonitor) + ', ' + \
		@@ 8349-8392 (lines=44) @@
8346		for _ in range(0, length):
8347		array.append(DiagnosticInfo.from_binary(data))
8348		obj.DiagnosticInfos = array
8349		return obj
8350
8351		def __str__(self):
8352		return 'ReadResponse(' + 'TypeId:' + str(self.TypeId) + ', ' + \
8353		'ResponseHeader:' + str(self.ResponseHeader) + ', ' + \
8354		'Results:' + str(self.Results) + ', ' + \
8355		'DiagnosticInfos:' + str(self.DiagnosticInfos) + ')'
8356
8357		__repr__ = __str__
8358
8359
8360		class HistoryReadValueId(FrozenClass):
8361		'''
8362		:ivar NodeId:
8363		:vartype NodeId: NodeId
8364		:ivar IndexRange:
8365		:vartype IndexRange: String
8366		:ivar DataEncoding:
8367		:vartype DataEncoding: QualifiedName
8368		:ivar ContinuationPoint:
8369		:vartype ContinuationPoint: ByteString
8370		'''
8371
8372		ua_types = [
8373		('NodeId', 'NodeId'),
8374		('IndexRange', 'String'),
8375		('DataEncoding', 'QualifiedName'),
8376		('ContinuationPoint', 'ByteString'),
8377		]
8378
8379		def __init__(self):
8380		self.NodeId = NodeId()
8381		self.IndexRange = None
8382		self.DataEncoding = QualifiedName()
8383		self.ContinuationPoint = None
8384		self._freeze = True
8385
8386		def to_binary(self):
8387		packet = []
8388		packet.append(self.NodeId.to_binary())
8389		packet.append(uabin.Primitives.String.pack(self.IndexRange))
8390		packet.append(self.DataEncoding.to_binary())
8391		packet.append(uabin.Primitives.ByteString.pack(self.ContinuationPoint))
8392		return b''.join(packet)
8393
8394		@staticmethod
8395		def from_binary(data):
		@@ 1918-1961 (lines=44) @@
1915
1916		ua_types = [
1917		('EndpointUrl', 'String'),
1918		('Server', 'ApplicationDescription'),
1919		('ServerCertificate', 'ByteString'),
1920		('SecurityMode', 'MessageSecurityMode'),
1921		('SecurityPolicyUri', 'String'),
1922		('UserIdentityTokens', 'ListOfUserTokenPolicy'),
1923		('TransportProfileUri', 'String'),
1924		('SecurityLevel', 'Byte'),
1925		]
1926
1927		def __init__(self):
1928		self.EndpointUrl = None
1929		self.Server = ApplicationDescription()
1930		self.ServerCertificate = None
1931		self.SecurityMode = MessageSecurityMode(0)
1932		self.SecurityPolicyUri = None
1933		self.UserIdentityTokens = []
1934		self.TransportProfileUri = None
1935		self.SecurityLevel = 0
1936		self._freeze = True
1937
1938		def to_binary(self):
1939		packet = []
1940		packet.append(uabin.Primitives.String.pack(self.EndpointUrl))
1941		packet.append(self.Server.to_binary())
1942		packet.append(uabin.Primitives.ByteString.pack(self.ServerCertificate))
1943		packet.append(uabin.Primitives.UInt32.pack(self.SecurityMode.value))
1944		packet.append(uabin.Primitives.String.pack(self.SecurityPolicyUri))
1945		packet.append(uabin.Primitives.Int32.pack(len(self.UserIdentityTokens)))
1946		for fieldname in self.UserIdentityTokens:
1947		packet.append(fieldname.to_binary())
1948		packet.append(uabin.Primitives.String.pack(self.TransportProfileUri))
1949		packet.append(uabin.Primitives.Byte.pack(self.SecurityLevel))
1950		return b''.join(packet)
1951
1952		@staticmethod
1953		def from_binary(data):
1954		obj = EndpointDescription()
1955		self.EndpointUrl = uabin.Primitives.String.unpack(data)
1956		obj.Server = ApplicationDescription.from_binary(data)
1957		self.ServerCertificate = uabin.Primitives.ByteString.unpack(data)
1958		self.SecurityMode = MessageSecurityMode(uabin.Primitives.UInt32.unpack(data))
1959		self.SecurityPolicyUri = uabin.Primitives.String.unpack(data)
1960		length = uabin.Primitives.Int32.unpack(data)
1961		array = []
1962		if length != -1:
1963		for _ in range(0, length):
1964		array.append(UserTokenPolicy.from_binary(data))
		@@ 7582-7625 (lines=44) @@
7579		('OperandDiagnosticInfos', 'ListOfDiagnosticInfo'),
7580		]
7581
7582		def __init__(self):
7583		self.StatusCode = StatusCode()
7584		self.OperandStatusCodes = []
7585		self.OperandDiagnosticInfos = []
7586		self._freeze = True
7587
7588		def to_binary(self):
7589		packet = []
7590		packet.append(self.StatusCode.to_binary())
7591		packet.append(uabin.Primitives.Int32.pack(len(self.OperandStatusCodes)))
7592		for fieldname in self.OperandStatusCodes:
7593		packet.append(fieldname.to_binary())
7594		packet.append(uabin.Primitives.Int32.pack(len(self.OperandDiagnosticInfos)))
7595		for fieldname in self.OperandDiagnosticInfos:
7596		packet.append(fieldname.to_binary())
7597		return b''.join(packet)
7598
7599		@staticmethod
7600		def from_binary(data):
7601		obj = ContentFilterElementResult()
7602		obj.StatusCode = StatusCode.from_binary(data)
7603		length = uabin.Primitives.Int32.unpack(data)
7604		array = []
7605		if length != -1:
7606		for _ in range(0, length):
7607		array.append(StatusCode.from_binary(data))
7608		obj.OperandStatusCodes = array
7609		length = uabin.Primitives.Int32.unpack(data)
7610		array = []
7611		if length != -1:
7612		for _ in range(0, length):
7613		array.append(DiagnosticInfo.from_binary(data))
7614		obj.OperandDiagnosticInfos = array
7615		return obj
7616
7617		def __str__(self):
7618		return 'ContentFilterElementResult(' + 'StatusCode:' + str(self.StatusCode) + ', ' + \
7619		'OperandStatusCodes:' + str(self.OperandStatusCodes) + ', ' + \
7620		'OperandDiagnosticInfos:' + str(self.OperandDiagnosticInfos) + ')'
7621
7622		__repr__ = __str__
7623
7624
7625		class ContentFilterResult(FrozenClass):
7626		'''
7627		:ivar ElementResults:
7628		:vartype ElementResults: ContentFilterElementResult
		@@ 2567-2610 (lines=44) @@
2564		def to_binary(self):
2565		packet = []
2566		packet.append(uabin.Primitives.UInt32.pack(self.ChannelId))
2567		packet.append(uabin.Primitives.UInt32.pack(self.TokenId))
2568		packet.append(uabin.Primitives.DateTime.pack(self.CreatedAt))
2569		packet.append(uabin.Primitives.UInt32.pack(self.RevisedLifetime))
2570		return b''.join(packet)
2571
2572		@staticmethod
2573		def from_binary(data):
2574		obj = ChannelSecurityToken()
2575		self.ChannelId = uabin.Primitives.UInt32.unpack(data)
2576		self.TokenId = uabin.Primitives.UInt32.unpack(data)
2577		self.CreatedAt = uabin.Primitives.DateTime.unpack(data)
2578		self.RevisedLifetime = uabin.Primitives.UInt32.unpack(data)
2579		return obj
2580
2581		def __str__(self):
2582		return 'ChannelSecurityToken(' + 'ChannelId:' + str(self.ChannelId) + ', ' + \
2583		'TokenId:' + str(self.TokenId) + ', ' + \
2584		'CreatedAt:' + str(self.CreatedAt) + ', ' + \
2585		'RevisedLifetime:' + str(self.RevisedLifetime) + ')'
2586
2587		__repr__ = __str__
2588
2589
2590		class OpenSecureChannelParameters(FrozenClass):
2591		'''
2592		:ivar ClientProtocolVersion:
2593		:vartype ClientProtocolVersion: UInt32
2594		:ivar RequestType:
2595		:vartype RequestType: SecurityTokenRequestType
2596		:ivar SecurityMode:
2597		:vartype SecurityMode: MessageSecurityMode
2598		:ivar ClientNonce:
2599		:vartype ClientNonce: ByteString
2600		:ivar RequestedLifetime:
2601		:vartype RequestedLifetime: UInt32
2602		'''
2603
2604		ua_types = [
2605		('ClientProtocolVersion', 'UInt32'),
2606		('RequestType', 'SecurityTokenRequestType'),
2607		('SecurityMode', 'MessageSecurityMode'),
2608		('ClientNonce', 'ByteString'),
2609		('RequestedLifetime', 'UInt32'),
2610		]
2611
2612		def __init__(self):
2613		self.ClientProtocolVersion = 0

FreeOpcUa / python-opcua

Code Duplication Length = 44-46 lines in 8 locations

opcua/ua/uaprotocol_auto.py 8 locations