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

Code Duplication Length = 41-46 lines in 12 locations

opcua/ua/uaprotocol_auto.py 12 locations


        ('ResponseHeader', 'ResponseHeader'),
        ('Parameters', 'CancelResult'),
               ]

    def __init__(self):
        self.TypeId = FourByteNodeId(ObjectIds.CancelResponse_Encoding_DefaultBinary)
        self.ResponseHeader = ResponseHeader()
        self.Parameters = CancelResult()
        self._freeze = True

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

    @staticmethod
    def from_binary(data):
        obj = CancelResponse()
        obj.TypeId = NodeId.from_binary(data)
        obj.ResponseHeader = ResponseHeader.from_binary(data)
        obj.Parameters = CancelResult.from_binary(data)
        return obj

    def __str__(self):
        return 'CancelResponse(' + 'TypeId:' + str(self.TypeId) + ', ' + \
               'ResponseHeader:' + str(self.ResponseHeader) + ', ' + \
               'Parameters:' + str(self.Parameters) + ')'

    __repr__ = __str__


class NodeAttributes(FrozenClass):
    '''
    The base attributes for all nodes.

    :ivar SpecifiedAttributes:
    :vartype SpecifiedAttributes: UInt32
    :ivar DisplayName:
    :vartype DisplayName: LocalizedText
    :ivar Description:
    :vartype Description: LocalizedText
    :ivar WriteMask:
    :vartype WriteMask: UInt32
    :ivar UserWriteMask:
    :vartype UserWriteMask: UInt32
    '''

    ua_types = [
        ('SpecifiedAttributes', 'UInt32'),
        ('DisplayName', 'LocalizedText'),

    :ivar ProductUri:
    :vartype ProductUri: String
    :ivar ApplicationName:
    :vartype ApplicationName: LocalizedText
    :ivar ApplicationType:
    :vartype ApplicationType: ApplicationType
    :ivar GatewayServerUri:
    :vartype GatewayServerUri: String
    :ivar DiscoveryProfileUri:
    :vartype DiscoveryProfileUri: String
    :ivar DiscoveryUrls:
    :vartype DiscoveryUrls: String
    '''

    ua_types = [
        ('ApplicationUri', 'String'),
        ('ProductUri', 'String'),
        ('ApplicationName', 'LocalizedText'),
        ('ApplicationType', 'ApplicationType'),
        ('GatewayServerUri', 'String'),
        ('DiscoveryProfileUri', 'String'),
        ('DiscoveryUrls', 'ListOfString'),
               ]

    def __init__(self):
        self.ApplicationUri = None
        self.ProductUri = None
        self.ApplicationName = LocalizedText()
        self.ApplicationType = ApplicationType(0)
        self.GatewayServerUri = None
        self.DiscoveryProfileUri = None
        self.DiscoveryUrls = []
        self._freeze = True

    def to_binary(self):
        packet = []
        packet.append(uabin.Primitives.String.pack(self.ApplicationUri))
        packet.append(uabin.Primitives.String.pack(self.ProductUri))
        packet.append(self.ApplicationName.to_binary())
        packet.append(uabin.Primitives.UInt32.pack(self.ApplicationType.value))
        packet.append(uabin.Primitives.String.pack(self.GatewayServerUri))
        packet.append(uabin.Primitives.String.pack(self.DiscoveryProfileUri))
        packet.append(uabin.Primitives.Int32.pack(len(self.DiscoveryUrls)))
        for fieldname in self.DiscoveryUrls:
            packet.append(uabin.Primitives.String.pack(fieldname))
        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        obj = ApplicationDescription()
        self.ApplicationUri = uabin.Primitives.String.unpack(data)
        self.ProductUri = uabin.Primitives.String.unpack(data)

        packet.append(uabin.Primitives.Int32.pack(len(self.LastMethodOutputArguments)))
        for fieldname in self.LastMethodOutputArguments:
            packet.append(fieldname.to_binary())
        packet.append(uabin.Primitives.DateTime.pack(self.LastMethodCallTime))
        packet.append(self.LastMethodReturnStatus.to_binary())
        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        obj = ProgramDiagnosticDataType()
        obj.CreateSessionId = NodeId.from_binary(data)
        self.CreateClientName = uabin.Primitives.String.unpack(data)
        self.InvocationCreationTime = uabin.Primitives.DateTime.unpack(data)
        self.LastTransitionTime = uabin.Primitives.DateTime.unpack(data)
        self.LastMethodCall = uabin.Primitives.String.unpack(data)
        obj.LastMethodSessionId = NodeId.from_binary(data)
        length = uabin.Primitives.Int32.unpack(data)
        array = []
        if length != -1:
            for _ in range(0, length):
                array.append(Argument.from_binary(data))
        obj.LastMethodInputArguments = array
        length = uabin.Primitives.Int32.unpack(data)
        array = []
        if length != -1:
            for _ in range(0, length):
                array.append(Argument.from_binary(data))
        obj.LastMethodOutputArguments = array
        self.LastMethodCallTime = uabin.Primitives.DateTime.unpack(data)
        obj.LastMethodReturnStatus = StatusResult.from_binary(data)
        return obj

    def __str__(self):
        return 'ProgramDiagnosticDataType(' + 'CreateSessionId:' + str(self.CreateSessionId) + ', ' + \
               'CreateClientName:' + str(self.CreateClientName) + ', ' + \
               'InvocationCreationTime:' + str(self.InvocationCreationTime) + ', ' + \
               'LastTransitionTime:' + str(self.LastTransitionTime) + ', ' + \
               'LastMethodCall:' + str(self.LastMethodCall) + ', ' + \
               'LastMethodSessionId:' + str(self.LastMethodSessionId) + ', ' + \
               'LastMethodInputArguments:' + str(self.LastMethodInputArguments) + ', ' + \
               'LastMethodOutputArguments:' + str(self.LastMethodOutputArguments) + ', ' + \
               'LastMethodCallTime:' + str(self.LastMethodCallTime) + ', ' + \
               'LastMethodReturnStatus:' + str(self.LastMethodReturnStatus) + ')'

    __repr__ = __str__


class Annotation(FrozenClass):
    '''
    :ivar Message:

    def from_binary(data):
        obj = MonitoredItemModifyResult()
        obj.StatusCode = StatusCode.from_binary(data)
        self.RevisedSamplingInterval = uabin.Primitives.Double.unpack(data)
        self.RevisedQueueSize = uabin.Primitives.UInt32.unpack(data)
        obj.FilterResult = uabin.extensionobject_from_binary(data)
        return obj

    def __str__(self):
        return 'MonitoredItemModifyResult(' + 'StatusCode:' + str(self.StatusCode) + ', ' + \
               'RevisedSamplingInterval:' + str(self.RevisedSamplingInterval) + ', ' + \
               'RevisedQueueSize:' + str(self.RevisedQueueSize) + ', ' + \
               'FilterResult:' + str(self.FilterResult) + ')'

    __repr__ = __str__


class ModifyMonitoredItemsParameters(FrozenClass):
    '''
    :ivar SubscriptionId:
    :vartype SubscriptionId: UInt32
    :ivar TimestampsToReturn:
    :vartype TimestampsToReturn: TimestampsToReturn
    :ivar ItemsToModify:
    :vartype ItemsToModify: MonitoredItemModifyRequest
    '''

    ua_types = [
        ('SubscriptionId', 'UInt32'),
        ('TimestampsToReturn', 'TimestampsToReturn'),
        ('ItemsToModify', 'ListOfMonitoredItemModifyRequest'),
               ]

    def __init__(self):
        self.SubscriptionId = 0
        self.TimestampsToReturn = TimestampsToReturn(0)
        self.ItemsToModify = []
        self._freeze = True

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

    @staticmethod
    def from_binary(data):

        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        obj = ContentFilter()
        length = uabin.Primitives.Int32.unpack(data)
        array = []
        if length != -1:
            for _ in range(0, length):
                array.append(ContentFilterElement.from_binary(data))
        obj.Elements = array
        return obj

    def __str__(self):
        return 'ContentFilter(' + 'Elements:' + str(self.Elements) + ')'

    __repr__ = __str__


class ElementOperand(FrozenClass):
    '''
    :ivar Index:
    :vartype Index: UInt32
    '''

    ua_types = [
        ('Index', 'UInt32'),
               ]

    def __init__(self):
        self.Index = 0
        self._freeze = True

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

    @staticmethod
    def from_binary(data):
        obj = ElementOperand()
        self.Index = uabin.Primitives.UInt32.unpack(data)
        return obj

    def __str__(self):
        return 'ElementOperand(' + 'Index:' + str(self.Index) + ')'

    __repr__ = __str__



        self._freeze = True

    def to_binary(self):
        packet = []
        packet.append(uabin.Primitives.ByteString.pack(self.CertificateData))
        packet.append(uabin.Primitives.ByteString.pack(self.Signature))
        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        obj = SignedSoftwareCertificate()
        self.CertificateData = uabin.Primitives.ByteString.unpack(data)
        self.Signature = uabin.Primitives.ByteString.unpack(data)
        return obj

    def __str__(self):
        return 'SignedSoftwareCertificate(' + 'CertificateData:' + str(self.CertificateData) + ', ' + \
               'Signature:' + str(self.Signature) + ')'

    __repr__ = __str__


class SignatureData(FrozenClass):
    '''
    A digital signature.

    :ivar Algorithm:
    :vartype Algorithm: String
    :ivar Signature:
    :vartype Signature: ByteString
    '''

    ua_types = [
        ('Algorithm', 'String'),
        ('Signature', 'ByteString'),
               ]

    def __init__(self):
        self.Algorithm = None
        self.Signature = None
        self._freeze = True

    def to_binary(self):
        packet = []
        packet.append(uabin.Primitives.String.pack(self.Algorithm))
        packet.append(uabin.Primitives.ByteString.pack(self.Signature))
        return b''.join(packet)

    @staticmethod
    def from_binary(data):


class BrowseParameters(FrozenClass):
    '''
    :ivar View:
    :vartype View: ViewDescription
    :ivar RequestedMaxReferencesPerNode:
    :vartype RequestedMaxReferencesPerNode: UInt32
    :ivar NodesToBrowse:
    :vartype NodesToBrowse: BrowseDescription
    '''

    ua_types = [
        ('View', 'ViewDescription'),
        ('RequestedMaxReferencesPerNode', 'UInt32'),
        ('NodesToBrowse', 'ListOfBrowseDescription'),
               ]

    def __init__(self):
        self.View = ViewDescription()
        self.RequestedMaxReferencesPerNode = 0
        self.NodesToBrowse = []
        self._freeze = True

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

    @staticmethod
    def from_binary(data):
        obj = BrowseParameters()
        obj.View = ViewDescription.from_binary(data)
        self.RequestedMaxReferencesPerNode = uabin.Primitives.UInt32.unpack(data)
        length = uabin.Primitives.Int32.unpack(data)
        array = []
        if length != -1:
            for _ in range(0, length):
                array.append(BrowseDescription.from_binary(data))
        obj.NodesToBrowse = array
        return obj

    def __str__(self):
        return 'BrowseParameters(' + 'View:' + str(self.View) + ', ' + \
               'RequestedMaxReferencesPerNode:' + str(self.RequestedMaxReferencesPerNode) + ', ' + \
               'NodesToBrowse:' + str(self.NodesToBrowse) + ')'

    __repr__ = __str__


        length = uabin.Primitives.Int32.unpack(data)
        array = []
        if length != -1:
            for _ in range(0, length):
                array.append(DiagnosticInfo.from_binary(data))
        obj.DiagnosticInfos = array
        return obj

    def __str__(self):
        return 'ActivateSessionResult(' + 'ServerNonce:' + str(self.ServerNonce) + ', ' + \
               'Results:' + str(self.Results) + ', ' + \
               'DiagnosticInfos:' + str(self.DiagnosticInfos) + ')'

    __repr__ = __str__


class ActivateSessionResponse(FrozenClass):
    '''
    Activates a session with the server.

    :ivar TypeId:
    :vartype TypeId: NodeId
    :ivar ResponseHeader:
    :vartype ResponseHeader: ResponseHeader
    :ivar Parameters:
    :vartype Parameters: ActivateSessionResult
    '''

    ua_types = [
        ('TypeId', 'NodeId'),
        ('ResponseHeader', 'ResponseHeader'),
        ('Parameters', 'ActivateSessionResult'),
               ]

    def __init__(self):
        self.TypeId = FourByteNodeId(ObjectIds.ActivateSessionResponse_Encoding_DefaultBinary)
        self.ResponseHeader = ResponseHeader()
        self.Parameters = ActivateSessionResult()
        self._freeze = True

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

    @staticmethod
    def from_binary(data):
        obj = ActivateSessionResponse()
        obj.TypeId = NodeId.from_binary(data)
        obj.ResponseHeader = ResponseHeader.from_binary(data)

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

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

    def __str__(self):
        return 'DataChangeNotification(' + 'MonitoredItems:' + str(self.MonitoredItems) + ', ' + \
               'DiagnosticInfos:' + str(self.DiagnosticInfos) + ')'

    __repr__ = __str__


class MonitoredItemNotification(FrozenClass):
    '''
    :ivar ClientHandle:
    :vartype ClientHandle: UInt32
    :ivar Value:
    :vartype Value: DataValue
    '''

    ua_types = [
        ('ClientHandle', 'UInt32'),
        ('Value', 'DataValue'),
               ]

    def __init__(self):
        self.ClientHandle = 0
        self.Value = DataValue()

        packet.append(uabin.Primitives.Int32.pack(len(self.DiagnosticInfos)))
        for fieldname in self.DiagnosticInfos:
            packet.append(fieldname.to_binary())
        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        obj = CreateMonitoredItemsResponse()
        obj.TypeId = NodeId.from_binary(data)
        obj.ResponseHeader = ResponseHeader.from_binary(data)
        length = uabin.Primitives.Int32.unpack(data)
        array = []
        if length != -1:
            for _ in range(0, length):
                array.append(MonitoredItemCreateResult.from_binary(data))
        obj.Results = array
        length = uabin.Primitives.Int32.unpack(data)
        array = []
        if length != -1:
            for _ in range(0, length):
                array.append(DiagnosticInfo.from_binary(data))
        obj.DiagnosticInfos = array
        return obj

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

    __repr__ = __str__


class MonitoredItemModifyRequest(FrozenClass):
    '''
    :ivar MonitoredItemId:
    :vartype MonitoredItemId: UInt32
    :ivar RequestedParameters:
    :vartype RequestedParameters: MonitoringParameters
    '''

    ua_types = [
        ('MonitoredItemId', 'UInt32'),
        ('RequestedParameters', 'MonitoringParameters'),
               ]

    def __init__(self):
        self.MonitoredItemId = 0
        self.RequestedParameters = MonitoringParameters()
        self._freeze = True

        self.DataValues = []
        self._freeze = True

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

    @staticmethod
    def from_binary(data):
        obj = HistoryData()
        length = uabin.Primitives.Int32.unpack(data)
        array = []
        if length != -1:
            for _ in range(0, length):
                array.append(DataValue.from_binary(data))
        obj.DataValues = array
        return obj

    def __str__(self):
        return 'HistoryData(' + 'DataValues:' + str(self.DataValues) + ')'

    __repr__ = __str__


class ModificationInfo(FrozenClass):
    '''
    :ivar ModificationTime:
    :vartype ModificationTime: DateTime
    :ivar UpdateType:
    :vartype UpdateType: HistoryUpdateType
    :ivar UserName:
    :vartype UserName: String
    '''

    ua_types = [
        ('ModificationTime', 'DateTime'),
        ('UpdateType', 'HistoryUpdateType'),
        ('UserName', 'String'),
               ]

    def __init__(self):
        self.ModificationTime = datetime.utcnow()
        self.UpdateType = HistoryUpdateType(0)
        self.UserName = None
        self._freeze = True

    def to_binary(self):

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

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

    __repr__ = __str__


class ChannelSecurityToken(FrozenClass):
    '''
    The token that identifies a set of keys for an active secure channel.

    :ivar ChannelId:
    :vartype ChannelId: UInt32
    :ivar TokenId:
    :vartype TokenId: UInt32
    :ivar CreatedAt:
    :vartype CreatedAt: DateTime
    :ivar RevisedLifetime:
    :vartype RevisedLifetime: UInt32
    '''

    ua_types = [
        ('ChannelId', 'UInt32'),
        ('TokenId', 'UInt32'),
        ('CreatedAt', 'DateTime'),
        ('RevisedLifetime', 'UInt32'),
               ]

    def __init__(self):
        self.ChannelId = 0
        self.TokenId = 0
        self.CreatedAt = datetime.utcnow()
        self.RevisedLifetime = 0
        self._freeze = True

    def to_binary(self):
        packet = []
        packet.append(uabin.Primitives.UInt32.pack(self.ChannelId))
        packet.append(uabin.Primitives.UInt32.pack(self.TokenId))

		@@ 3902-3947 (lines=46) @@
3899		('ResponseHeader', 'ResponseHeader'),
3900		('Parameters', 'CancelResult'),
3901		]
3902
3903		def __init__(self):
3904		self.TypeId = FourByteNodeId(ObjectIds.CancelResponse_Encoding_DefaultBinary)
3905		self.ResponseHeader = ResponseHeader()
3906		self.Parameters = CancelResult()
3907		self._freeze = True
3908
3909		def to_binary(self):
3910		packet = []
3911		packet.append(self.TypeId.to_binary())
3912		packet.append(self.ResponseHeader.to_binary())
3913		packet.append(self.Parameters.to_binary())
3914		return b''.join(packet)
3915
3916		@staticmethod
3917		def from_binary(data):
3918		obj = CancelResponse()
3919		obj.TypeId = NodeId.from_binary(data)
3920		obj.ResponseHeader = ResponseHeader.from_binary(data)
3921		obj.Parameters = CancelResult.from_binary(data)
3922		return obj
3923
3924		def __str__(self):
3925		return 'CancelResponse(' + 'TypeId:' + str(self.TypeId) + ', ' + \
3926		'ResponseHeader:' + str(self.ResponseHeader) + ', ' + \
3927		'Parameters:' + str(self.Parameters) + ')'
3928
3929		__repr__ = __str__
3930
3931
3932		class NodeAttributes(FrozenClass):
3933		'''
3934		The base attributes for all nodes.
3935
3936		:ivar SpecifiedAttributes:
3937		:vartype SpecifiedAttributes: UInt32
3938		:ivar DisplayName:
3939		:vartype DisplayName: LocalizedText
3940		:ivar Description:
3941		:vartype Description: LocalizedText
3942		:ivar WriteMask:
3943		:vartype WriteMask: UInt32
3944		:ivar UserWriteMask:
3945		:vartype UserWriteMask: UInt32
3946		'''
3947
3948		ua_types = [
3949		('SpecifiedAttributes', 'UInt32'),
3950		('DisplayName', 'LocalizedText'),
		@@ 1194-1239 (lines=46) @@
1191		:ivar ProductUri:
1192		:vartype ProductUri: String
1193		:ivar ApplicationName:
1194		:vartype ApplicationName: LocalizedText
1195		:ivar ApplicationType:
1196		:vartype ApplicationType: ApplicationType
1197		:ivar GatewayServerUri:
1198		:vartype GatewayServerUri: String
1199		:ivar DiscoveryProfileUri:
1200		:vartype DiscoveryProfileUri: String
1201		:ivar DiscoveryUrls:
1202		:vartype DiscoveryUrls: String
1203		'''
1204
1205		ua_types = [
1206		('ApplicationUri', 'String'),
1207		('ProductUri', 'String'),
1208		('ApplicationName', 'LocalizedText'),
1209		('ApplicationType', 'ApplicationType'),
1210		('GatewayServerUri', 'String'),
1211		('DiscoveryProfileUri', 'String'),
1212		('DiscoveryUrls', 'ListOfString'),
1213		]
1214
1215		def __init__(self):
1216		self.ApplicationUri = None
1217		self.ProductUri = None
1218		self.ApplicationName = LocalizedText()
1219		self.ApplicationType = ApplicationType(0)
1220		self.GatewayServerUri = None
1221		self.DiscoveryProfileUri = None
1222		self.DiscoveryUrls = []
1223		self._freeze = True
1224
1225		def to_binary(self):
1226		packet = []
1227		packet.append(uabin.Primitives.String.pack(self.ApplicationUri))
1228		packet.append(uabin.Primitives.String.pack(self.ProductUri))
1229		packet.append(self.ApplicationName.to_binary())
1230		packet.append(uabin.Primitives.UInt32.pack(self.ApplicationType.value))
1231		packet.append(uabin.Primitives.String.pack(self.GatewayServerUri))
1232		packet.append(uabin.Primitives.String.pack(self.DiscoveryProfileUri))
1233		packet.append(uabin.Primitives.Int32.pack(len(self.DiscoveryUrls)))
1234		for fieldname in self.DiscoveryUrls:
1235		packet.append(uabin.Primitives.String.pack(fieldname))
1236		return b''.join(packet)
1237
1238		@staticmethod
1239		def from_binary(data):
1240		obj = ApplicationDescription()
1241		self.ApplicationUri = uabin.Primitives.String.unpack(data)
1242		self.ProductUri = uabin.Primitives.String.unpack(data)
		@@ 14770-14813 (lines=44) @@
14767		packet.append(uabin.Primitives.Int32.pack(len(self.LastMethodOutputArguments)))
14768		for fieldname in self.LastMethodOutputArguments:
14769		packet.append(fieldname.to_binary())
14770		packet.append(uabin.Primitives.DateTime.pack(self.LastMethodCallTime))
14771		packet.append(self.LastMethodReturnStatus.to_binary())
14772		return b''.join(packet)
14773
14774		@staticmethod
14775		def from_binary(data):
14776		obj = ProgramDiagnosticDataType()
14777		obj.CreateSessionId = NodeId.from_binary(data)
14778		self.CreateClientName = uabin.Primitives.String.unpack(data)
14779		self.InvocationCreationTime = uabin.Primitives.DateTime.unpack(data)
14780		self.LastTransitionTime = uabin.Primitives.DateTime.unpack(data)
14781		self.LastMethodCall = uabin.Primitives.String.unpack(data)
14782		obj.LastMethodSessionId = NodeId.from_binary(data)
14783		length = uabin.Primitives.Int32.unpack(data)
14784		array = []
14785		if length != -1:
14786		for _ in range(0, length):
14787		array.append(Argument.from_binary(data))
14788		obj.LastMethodInputArguments = array
14789		length = uabin.Primitives.Int32.unpack(data)
14790		array = []
14791		if length != -1:
14792		for _ in range(0, length):
14793		array.append(Argument.from_binary(data))
14794		obj.LastMethodOutputArguments = array
14795		self.LastMethodCallTime = uabin.Primitives.DateTime.unpack(data)
14796		obj.LastMethodReturnStatus = StatusResult.from_binary(data)
14797		return obj
14798
14799		def __str__(self):
14800		return 'ProgramDiagnosticDataType(' + 'CreateSessionId:' + str(self.CreateSessionId) + ', ' + \
14801		'CreateClientName:' + str(self.CreateClientName) + ', ' + \
14802		'InvocationCreationTime:' + str(self.InvocationCreationTime) + ', ' + \
14803		'LastTransitionTime:' + str(self.LastTransitionTime) + ', ' + \
14804		'LastMethodCall:' + str(self.LastMethodCall) + ', ' + \
14805		'LastMethodSessionId:' + str(self.LastMethodSessionId) + ', ' + \
14806		'LastMethodInputArguments:' + str(self.LastMethodInputArguments) + ', ' + \
14807		'LastMethodOutputArguments:' + str(self.LastMethodOutputArguments) + ', ' + \
14808		'LastMethodCallTime:' + str(self.LastMethodCallTime) + ', ' + \
14809		'LastMethodReturnStatus:' + str(self.LastMethodReturnStatus) + ')'
14810
14811		__repr__ = __str__
14812
14813
14814		class Annotation(FrozenClass):
14815		'''
14816		:ivar Message:
		@@ 10808-10851 (lines=44) @@
10805		def from_binary(data):
10806		obj = MonitoredItemModifyResult()
10807		obj.StatusCode = StatusCode.from_binary(data)
10808		self.RevisedSamplingInterval = uabin.Primitives.Double.unpack(data)
10809		self.RevisedQueueSize = uabin.Primitives.UInt32.unpack(data)
10810		obj.FilterResult = uabin.extensionobject_from_binary(data)
10811		return obj
10812
10813		def __str__(self):
10814		return 'MonitoredItemModifyResult(' + 'StatusCode:' + str(self.StatusCode) + ', ' + \
10815		'RevisedSamplingInterval:' + str(self.RevisedSamplingInterval) + ', ' + \
10816		'RevisedQueueSize:' + str(self.RevisedQueueSize) + ', ' + \
10817		'FilterResult:' + str(self.FilterResult) + ')'
10818
10819		__repr__ = __str__
10820
10821
10822		class ModifyMonitoredItemsParameters(FrozenClass):
10823		'''
10824		:ivar SubscriptionId:
10825		:vartype SubscriptionId: UInt32
10826		:ivar TimestampsToReturn:
10827		:vartype TimestampsToReturn: TimestampsToReturn
10828		:ivar ItemsToModify:
10829		:vartype ItemsToModify: MonitoredItemModifyRequest
10830		'''
10831
10832		ua_types = [
10833		('SubscriptionId', 'UInt32'),
10834		('TimestampsToReturn', 'TimestampsToReturn'),
10835		('ItemsToModify', 'ListOfMonitoredItemModifyRequest'),
10836		]
10837
10838		def __init__(self):
10839		self.SubscriptionId = 0
10840		self.TimestampsToReturn = TimestampsToReturn(0)
10841		self.ItemsToModify = []
10842		self._freeze = True
10843
10844		def to_binary(self):
10845		packet = []
10846		packet.append(uabin.Primitives.UInt32.pack(self.SubscriptionId))
10847		packet.append(uabin.Primitives.UInt32.pack(self.TimestampsToReturn.value))
10848		packet.append(uabin.Primitives.Int32.pack(len(self.ItemsToModify)))
10849		for fieldname in self.ItemsToModify:
10850		packet.append(fieldname.to_binary())
10851		return b''.join(packet)
10852
10853		@staticmethod
10854		def from_binary(data):
		@@ 7370-7413 (lines=44) @@
7367		return b''.join(packet)
7368
7369		@staticmethod
7370		def from_binary(data):
7371		obj = ContentFilter()
7372		length = uabin.Primitives.Int32.unpack(data)
7373		array = []
7374		if length != -1:
7375		for _ in range(0, length):
7376		array.append(ContentFilterElement.from_binary(data))
7377		obj.Elements = array
7378		return obj
7379
7380		def __str__(self):
7381		return 'ContentFilter(' + 'Elements:' + str(self.Elements) + ')'
7382
7383		__repr__ = __str__
7384
7385
7386		class ElementOperand(FrozenClass):
7387		'''
7388		:ivar Index:
7389		:vartype Index: UInt32
7390		'''
7391
7392		ua_types = [
7393		('Index', 'UInt32'),
7394		]
7395
7396		def __init__(self):
7397		self.Index = 0
7398		self._freeze = True
7399
7400		def to_binary(self):
7401		packet = []
7402		packet.append(uabin.Primitives.UInt32.pack(self.Index))
7403		return b''.join(packet)
7404
7405		@staticmethod
7406		def from_binary(data):
7407		obj = ElementOperand()
7408		self.Index = uabin.Primitives.UInt32.unpack(data)
7409		return obj
7410
7411		def __str__(self):
7412		return 'ElementOperand(' + 'Index:' + str(self.Index) + ')'
7413
7414		__repr__ = __str__
7415
7416
		@@ 2889-2932 (lines=44) @@
2886		self._freeze = True
2887
2888		def to_binary(self):
2889		packet = []
2890		packet.append(uabin.Primitives.ByteString.pack(self.CertificateData))
2891		packet.append(uabin.Primitives.ByteString.pack(self.Signature))
2892		return b''.join(packet)
2893
2894		@staticmethod
2895		def from_binary(data):
2896		obj = SignedSoftwareCertificate()
2897		self.CertificateData = uabin.Primitives.ByteString.unpack(data)
2898		self.Signature = uabin.Primitives.ByteString.unpack(data)
2899		return obj
2900
2901		def __str__(self):
2902		return 'SignedSoftwareCertificate(' + 'CertificateData:' + str(self.CertificateData) + ', ' + \
2903		'Signature:' + str(self.Signature) + ')'
2904
2905		__repr__ = __str__
2906
2907
2908		class SignatureData(FrozenClass):
2909		'''
2910		A digital signature.
2911
2912		:ivar Algorithm:
2913		:vartype Algorithm: String
2914		:ivar Signature:
2915		:vartype Signature: ByteString
2916		'''
2917
2918		ua_types = [
2919		('Algorithm', 'String'),
2920		('Signature', 'ByteString'),
2921		]
2922
2923		def __init__(self):
2924		self.Algorithm = None
2925		self.Signature = None
2926		self._freeze = True
2927
2928		def to_binary(self):
2929		packet = []
2930		packet.append(uabin.Primitives.String.pack(self.Algorithm))
2931		packet.append(uabin.Primitives.ByteString.pack(self.Signature))
2932		return b''.join(packet)
2933
2934		@staticmethod
2935		def from_binary(data):
		@@ 5819-5864 (lines=46) @@
5816
5817		class BrowseParameters(FrozenClass):
5818		'''
5819		:ivar View:
5820		:vartype View: ViewDescription
5821		:ivar RequestedMaxReferencesPerNode:
5822		:vartype RequestedMaxReferencesPerNode: UInt32
5823		:ivar NodesToBrowse:
5824		:vartype NodesToBrowse: BrowseDescription
5825		'''
5826
5827		ua_types = [
5828		('View', 'ViewDescription'),
5829		('RequestedMaxReferencesPerNode', 'UInt32'),
5830		('NodesToBrowse', 'ListOfBrowseDescription'),
5831		]
5832
5833		def __init__(self):
5834		self.View = ViewDescription()
5835		self.RequestedMaxReferencesPerNode = 0
5836		self.NodesToBrowse = []
5837		self._freeze = True
5838
5839		def to_binary(self):
5840		packet = []
5841		packet.append(self.View.to_binary())
5842		packet.append(uabin.Primitives.UInt32.pack(self.RequestedMaxReferencesPerNode))
5843		packet.append(uabin.Primitives.Int32.pack(len(self.NodesToBrowse)))
5844		for fieldname in self.NodesToBrowse:
5845		packet.append(fieldname.to_binary())
5846		return b''.join(packet)
5847
5848		@staticmethod
5849		def from_binary(data):
5850		obj = BrowseParameters()
5851		obj.View = ViewDescription.from_binary(data)
5852		self.RequestedMaxReferencesPerNode = uabin.Primitives.UInt32.unpack(data)
5853		length = uabin.Primitives.Int32.unpack(data)
5854		array = []
5855		if length != -1:
5856		for _ in range(0, length):
5857		array.append(BrowseDescription.from_binary(data))
5858		obj.NodesToBrowse = array
5859		return obj
5860
5861		def __str__(self):
5862		return 'BrowseParameters(' + 'View:' + str(self.View) + ', ' + \
5863		'RequestedMaxReferencesPerNode:' + str(self.RequestedMaxReferencesPerNode) + ', ' + \
5864		'NodesToBrowse:' + str(self.NodesToBrowse) + ')'
5865
5866		__repr__ = __str__
5867
		@@ 3629-3674 (lines=46) @@
3626		length = uabin.Primitives.Int32.unpack(data)
3627		array = []
3628		if length != -1:
3629		for _ in range(0, length):
3630		array.append(DiagnosticInfo.from_binary(data))
3631		obj.DiagnosticInfos = array
3632		return obj
3633
3634		def __str__(self):
3635		return 'ActivateSessionResult(' + 'ServerNonce:' + str(self.ServerNonce) + ', ' + \
3636		'Results:' + str(self.Results) + ', ' + \
3637		'DiagnosticInfos:' + str(self.DiagnosticInfos) + ')'
3638
3639		__repr__ = __str__
3640
3641
3642		class ActivateSessionResponse(FrozenClass):
3643		'''
3644		Activates a session with the server.
3645
3646		:ivar TypeId:
3647		:vartype TypeId: NodeId
3648		:ivar ResponseHeader:
3649		:vartype ResponseHeader: ResponseHeader
3650		:ivar Parameters:
3651		:vartype Parameters: ActivateSessionResult
3652		'''
3653
3654		ua_types = [
3655		('TypeId', 'NodeId'),
3656		('ResponseHeader', 'ResponseHeader'),
3657		('Parameters', 'ActivateSessionResult'),
3658		]
3659
3660		def __init__(self):
3661		self.TypeId = FourByteNodeId(ObjectIds.ActivateSessionResponse_Encoding_DefaultBinary)
3662		self.ResponseHeader = ResponseHeader()
3663		self.Parameters = ActivateSessionResult()
3664		self._freeze = True
3665
3666		def to_binary(self):
3667		packet = []
3668		packet.append(self.TypeId.to_binary())
3669		packet.append(self.ResponseHeader.to_binary())
3670		packet.append(self.Parameters.to_binary())
3671		return b''.join(packet)
3672
3673		@staticmethod
3674		def from_binary(data):
3675		obj = ActivateSessionResponse()
3676		obj.TypeId = NodeId.from_binary(data)
3677		obj.ResponseHeader = ResponseHeader.from_binary(data)
		@@ 12240-12283 (lines=44) @@
12237		def to_binary(self):
12238		packet = []
12239		packet.append(uabin.Primitives.Int32.pack(len(self.MonitoredItems)))
12240		for fieldname in self.MonitoredItems:
12241		packet.append(fieldname.to_binary())
12242		packet.append(uabin.Primitives.Int32.pack(len(self.DiagnosticInfos)))
12243		for fieldname in self.DiagnosticInfos:
12244		packet.append(fieldname.to_binary())
12245		return b''.join(packet)
12246
12247		@staticmethod
12248		def from_binary(data):
12249		obj = DataChangeNotification()
12250		length = uabin.Primitives.Int32.unpack(data)
12251		array = []
12252		if length != -1:
12253		for _ in range(0, length):
12254		array.append(MonitoredItemNotification.from_binary(data))
12255		obj.MonitoredItems = array
12256		length = uabin.Primitives.Int32.unpack(data)
12257		array = []
12258		if length != -1:
12259		for _ in range(0, length):
12260		array.append(DiagnosticInfo.from_binary(data))
12261		obj.DiagnosticInfos = array
12262		return obj
12263
12264		def __str__(self):
12265		return 'DataChangeNotification(' + 'MonitoredItems:' + str(self.MonitoredItems) + ', ' + \
12266		'DiagnosticInfos:' + str(self.DiagnosticInfos) + ')'
12267
12268		__repr__ = __str__
12269
12270
12271		class MonitoredItemNotification(FrozenClass):
12272		'''
12273		:ivar ClientHandle:
12274		:vartype ClientHandle: UInt32
12275		:ivar Value:
12276		:vartype Value: DataValue
12277		'''
12278
12279		ua_types = [
12280		('ClientHandle', 'UInt32'),
12281		('Value', 'DataValue'),
12282		]
12283
12284		def __init__(self):
12285		self.ClientHandle = 0
12286		self.Value = DataValue()
		@@ 10702-10745 (lines=44) @@
10699		packet.append(uabin.Primitives.Int32.pack(len(self.DiagnosticInfos)))
10700		for fieldname in self.DiagnosticInfos:
10701		packet.append(fieldname.to_binary())
10702		return b''.join(packet)
10703
10704		@staticmethod
10705		def from_binary(data):
10706		obj = CreateMonitoredItemsResponse()
10707		obj.TypeId = NodeId.from_binary(data)
10708		obj.ResponseHeader = ResponseHeader.from_binary(data)
10709		length = uabin.Primitives.Int32.unpack(data)
10710		array = []
10711		if length != -1:
10712		for _ in range(0, length):
10713		array.append(MonitoredItemCreateResult.from_binary(data))
10714		obj.Results = array
10715		length = uabin.Primitives.Int32.unpack(data)
10716		array = []
10717		if length != -1:
10718		for _ in range(0, length):
10719		array.append(DiagnosticInfo.from_binary(data))
10720		obj.DiagnosticInfos = array
10721		return obj
10722
10723		def __str__(self):
10724		return 'CreateMonitoredItemsResponse(' + 'TypeId:' + str(self.TypeId) + ', ' + \
10725		'ResponseHeader:' + str(self.ResponseHeader) + ', ' + \
10726		'Results:' + str(self.Results) + ', ' + \
10727		'DiagnosticInfos:' + str(self.DiagnosticInfos) + ')'
10728
10729		__repr__ = __str__
10730
10731
10732		class MonitoredItemModifyRequest(FrozenClass):
10733		'''
10734		:ivar MonitoredItemId:
10735		:vartype MonitoredItemId: UInt32
10736		:ivar RequestedParameters:
10737		:vartype RequestedParameters: MonitoringParameters
10738		'''
10739
10740		ua_types = [
10741		('MonitoredItemId', 'UInt32'),
10742		('RequestedParameters', 'MonitoringParameters'),
10743		]
10744
10745		def __init__(self):
10746		self.MonitoredItemId = 0
10747		self.RequestedParameters = MonitoringParameters()
10748		self._freeze = True
		@@ 8713-8756 (lines=44) @@
8710		self.DataValues = []
8711		self._freeze = True
8712
8713		def to_binary(self):
8714		packet = []
8715		packet.append(uabin.Primitives.Int32.pack(len(self.DataValues)))
8716		for fieldname in self.DataValues:
8717		packet.append(fieldname.to_binary())
8718		return b''.join(packet)
8719
8720		@staticmethod
8721		def from_binary(data):
8722		obj = HistoryData()
8723		length = uabin.Primitives.Int32.unpack(data)
8724		array = []
8725		if length != -1:
8726		for _ in range(0, length):
8727		array.append(DataValue.from_binary(data))
8728		obj.DataValues = array
8729		return obj
8730
8731		def __str__(self):
8732		return 'HistoryData(' + 'DataValues:' + str(self.DataValues) + ')'
8733
8734		__repr__ = __str__
8735
8736
8737		class ModificationInfo(FrozenClass):
8738		'''
8739		:ivar ModificationTime:
8740		:vartype ModificationTime: DateTime
8741		:ivar UpdateType:
8742		:vartype UpdateType: HistoryUpdateType
8743		:ivar UserName:
8744		:vartype UserName: String
8745		'''
8746
8747		ua_types = [
8748		('ModificationTime', 'DateTime'),
8749		('UpdateType', 'HistoryUpdateType'),
8750		('UserName', 'String'),
8751		]
8752
8753		def __init__(self):
8754		self.ModificationTime = datetime.utcnow()
8755		self.UpdateType = HistoryUpdateType(0)
8756		self.UserName = None
8757		self._freeze = True
8758
8759		def to_binary(self):
		@@ 2524-2564 (lines=41) @@
2521		if length != -1:
2522		for _ in range(0, length):
2523		array.append(DiagnosticInfo.from_binary(data))
2524		obj.DiagnosticInfos = array
2525		return obj
2526
2527		def __str__(self):
2528		return 'RegisterServer2Response(' + 'TypeId:' + str(self.TypeId) + ', ' + \
2529		'ResponseHeader:' + str(self.ResponseHeader) + ', ' + \
2530		'ConfigurationResults:' + str(self.ConfigurationResults) + ', ' + \
2531		'DiagnosticInfos:' + str(self.DiagnosticInfos) + ')'
2532
2533		__repr__ = __str__
2534
2535
2536		class ChannelSecurityToken(FrozenClass):
2537		'''
2538		The token that identifies a set of keys for an active secure channel.
2539
2540		:ivar ChannelId:
2541		:vartype ChannelId: UInt32
2542		:ivar TokenId:
2543		:vartype TokenId: UInt32
2544		:ivar CreatedAt:
2545		:vartype CreatedAt: DateTime
2546		:ivar RevisedLifetime:
2547		:vartype RevisedLifetime: UInt32
2548		'''
2549
2550		ua_types = [
2551		('ChannelId', 'UInt32'),
2552		('TokenId', 'UInt32'),
2553		('CreatedAt', 'DateTime'),
2554		('RevisedLifetime', 'UInt32'),
2555		]
2556
2557		def __init__(self):
2558		self.ChannelId = 0
2559		self.TokenId = 0
2560		self.CreatedAt = datetime.utcnow()
2561		self.RevisedLifetime = 0
2562		self._freeze = True
2563
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))

FreeOpcUa / python-opcua

Code Duplication Length = 41-46 lines in 12 locations

opcua/ua/uaprotocol_auto.py 12 locations