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

Code Duplication Length = 41-48 lines in 12 locations

opcua/ua/uaprotocol_auto.py 12 locations



    __repr__ = __str__


class BrowsePathResult(FrozenClass):
    '''
    The result of a translate opearation.

    :ivar StatusCode:
    :vartype StatusCode: StatusCode
    :ivar Targets:
    :vartype Targets: BrowsePathTarget
    '''
    def __init__(self, binary=None):
        if binary is not None:
            self._binary_init(binary)
            self._freeze = True
            return
        self.StatusCode = StatusCode()
        self.Targets = []
        self._freeze = True

    def to_binary(self):
        packet = []
        packet.append(self.StatusCode.to_binary())
        packet.append(uatype_Int32.pack(len(self.Targets)))
        for fieldname in self.Targets:
            packet.append(fieldname.to_binary())
        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        return BrowsePathResult(data)

    def _binary_init(self, data):
        self.StatusCode = StatusCode.from_binary(data)
        length = uatype_Int32.unpack(data.read(4))[0]
        array = []
        if length != -1:
            for _ in range(0, length):
                array.append(BrowsePathTarget.from_binary(data))
        self.Targets = array

    def __str__(self):
        return 'BrowsePathResult(' + 'StatusCode:' + str(self.StatusCode) + ', ' + \
               'Targets:' + str(self.Targets) + ')'

    __repr__ = __str__




    __repr__ = __str__


class SetMonitoringModeParameters(FrozenClass):
    '''
    :ivar SubscriptionId:
    :vartype SubscriptionId: UInt32
    :ivar MonitoringMode:
    :vartype MonitoringMode: MonitoringMode
    :ivar MonitoredItemIds:
    :vartype MonitoredItemIds: UInt32
    '''
    def __init__(self, binary=None):
        if binary is not None:
            self._binary_init(binary)
            self._freeze = True
            return
        self.SubscriptionId = 0
        self.MonitoringMode = MonitoringMode(0)
        self.MonitoredItemIds = []
        self._freeze = True

    def to_binary(self):
        packet = []
        packet.append(uatype_UInt32.pack(self.SubscriptionId))
        packet.append(uatype_UInt32.pack(self.MonitoringMode.value))
        packet.append(uatype_Int32.pack(len(self.MonitoredItemIds)))
        for fieldname in self.MonitoredItemIds:
            packet.append(uatype_UInt32.pack(fieldname))
        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        return SetMonitoringModeParameters(data)

    def _binary_init(self, data):
        self.SubscriptionId = uatype_UInt32.unpack(data.read(4))[0]
        self.MonitoringMode = MonitoringMode(uatype_UInt32.unpack(data.read(4))[0])
        self.MonitoredItemIds = unpack_uatype_array('UInt32', data)

    def __str__(self):
        return 'SetMonitoringModeParameters(' + 'SubscriptionId:' + str(self.SubscriptionId) + ', ' + \
               'MonitoringMode:' + str(self.MonitoringMode) + ', ' + \
               'MonitoredItemIds:' + str(self.MonitoredItemIds) + ')'

    __repr__ = __str__




    __repr__ = __str__


class EventFieldList(FrozenClass):
    '''
    :ivar ClientHandle:
    :vartype ClientHandle: UInt32
    :ivar EventFields:
    :vartype EventFields: Variant
    '''
    def __init__(self, binary=None):
        if binary is not None:
            self._binary_init(binary)
            self._freeze = True
            return
        self.ClientHandle = 0
        self.EventFields = []
        self._freeze = True

    def to_binary(self):
        packet = []
        packet.append(uatype_UInt32.pack(self.ClientHandle))
        packet.append(uatype_Int32.pack(len(self.EventFields)))
        for fieldname in self.EventFields:
            packet.append(fieldname.to_binary())
        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        return EventFieldList(data)

    def _binary_init(self, data):
        self.ClientHandle = uatype_UInt32.unpack(data.read(4))[0]
        length = uatype_Int32.unpack(data.read(4))[0]
        array = []
        if length != -1:
            for _ in range(0, length):
                array.append(Variant.from_binary(data))
        self.EventFields = array

    def __str__(self):
        return 'EventFieldList(' + 'ClientHandle:' + str(self.ClientHandle) + ', ' + \
               'EventFields:' + str(self.EventFields) + ')'

    __repr__ = __str__




    __repr__ = __str__


class EventFilter(FrozenClass):
    '''
    :ivar SelectClauses:
    :vartype SelectClauses: SimpleAttributeOperand
    :ivar WhereClause:
    :vartype WhereClause: ContentFilter
    '''
    def __init__(self, binary=None):
        if binary is not None:
            self._binary_init(binary)
            self._freeze = True
            return
        self.SelectClauses = []
        self.WhereClause = ContentFilter()
        self._freeze = True

    def to_binary(self):
        packet = []
        packet.append(uatype_Int32.pack(len(self.SelectClauses)))
        for fieldname in self.SelectClauses:
            packet.append(fieldname.to_binary())
        packet.append(self.WhereClause.to_binary())
        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        return EventFilter(data)

    def _binary_init(self, data):
        length = uatype_Int32.unpack(data.read(4))[0]
        array = []
        if length != -1:
            for _ in range(0, length):
                array.append(SimpleAttributeOperand.from_binary(data))
        self.SelectClauses = array
        self.WhereClause = ContentFilter.from_binary(data)

    def __str__(self):
        return 'EventFilter(' + 'SelectClauses:' + str(self.SelectClauses) + ', ' + \
               'WhereClause:' + str(self.WhereClause) + ')'

    __repr__ = __str__




    __repr__ = __str__


class ContentFilterElement(FrozenClass):
    '''
    :ivar FilterOperator:
    :vartype FilterOperator: FilterOperator
    :ivar FilterOperands:
    :vartype FilterOperands: ExtensionObject
    '''
    def __init__(self, binary=None):
        if binary is not None:
            self._binary_init(binary)
            self._freeze = True
            return
        self.FilterOperator = FilterOperator(0)
        self.FilterOperands = []
        self._freeze = True

    def to_binary(self):
        packet = []
        packet.append(uatype_UInt32.pack(self.FilterOperator.value))
        packet.append(uatype_Int32.pack(len(self.FilterOperands)))
        for fieldname in self.FilterOperands:
            packet.append(extensionobject_to_binary(fieldname))
        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        return ContentFilterElement(data)

    def _binary_init(self, data):
        self.FilterOperator = FilterOperator(uatype_UInt32.unpack(data.read(4))[0])
        length = uatype_Int32.unpack(data.read(4))[0]
        array = []
        if length != -1:
            for _ in range(0, length):
                array.append(extensionobject_from_binary(data))
        self.FilterOperands = array

    def __str__(self):
        return 'ContentFilterElement(' + 'FilterOperator:' + str(self.FilterOperator) + ', ' + \
               'FilterOperands:' + str(self.FilterOperands) + ')'

    __repr__ = __str__




    __repr__ = __str__


class RegisterServer2Parameters(FrozenClass):
    '''
    :ivar Server:
    :vartype Server: RegisteredServer
    :ivar DiscoveryConfiguration:
    :vartype DiscoveryConfiguration: ExtensionObject
    '''
    def __init__(self, binary=None):
        if binary is not None:
            self._binary_init(binary)
            self._freeze = True
            return
        self.Server = RegisteredServer()
        self.DiscoveryConfiguration = []
        self._freeze = True

    def to_binary(self):
        packet = []
        packet.append(self.Server.to_binary())
        packet.append(uatype_Int32.pack(len(self.DiscoveryConfiguration)))
        for fieldname in self.DiscoveryConfiguration:
            packet.append(extensionobject_to_binary(fieldname))
        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        return RegisterServer2Parameters(data)

    def _binary_init(self, data):
        self.Server = RegisteredServer.from_binary(data)
        length = uatype_Int32.unpack(data.read(4))[0]
        array = []
        if length != -1:
            for _ in range(0, length):
                array.append(extensionobject_from_binary(data))
        self.DiscoveryConfiguration = array

    def __str__(self):
        return 'RegisterServer2Parameters(' + 'Server:' + str(self.Server) + ', ' + \
               'DiscoveryConfiguration:' + str(self.DiscoveryConfiguration) + ')'

    __repr__ = __str__




    __repr__ = __str__


class NotificationMessage(FrozenClass):
    '''
    :ivar SequenceNumber:
    :vartype SequenceNumber: UInt32
    :ivar PublishTime:
    :vartype PublishTime: DateTime
    :ivar NotificationData:
    :vartype NotificationData: ExtensionObject
    '''
    def __init__(self, binary=None):
        if binary is not None:
            self._binary_init(binary)
            self._freeze = True
            return
        self.SequenceNumber = 0
        self.PublishTime = datetime.now()
        self.NotificationData = []
        self._freeze = True

    def to_binary(self):
        packet = []
        packet.append(uatype_UInt32.pack(self.SequenceNumber))
        packet.append(pack_datetime(self.PublishTime))
        packet.append(uatype_Int32.pack(len(self.NotificationData)))
        for fieldname in self.NotificationData:
            packet.append(extensionobject_to_binary(fieldname))
        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        return NotificationMessage(data)

    def _binary_init(self, data):
        self.SequenceNumber = uatype_UInt32.unpack(data.read(4))[0]
        self.PublishTime = unpack_datetime(data)
        length = uatype_Int32.unpack(data.read(4))[0]
        array = []
        if length != -1:
            for _ in range(0, length):
                array.append(extensionobject_from_binary(data))
        self.NotificationData = array

    def __str__(self):
        return 'NotificationMessage(' + 'SequenceNumber:' + str(self.SequenceNumber) + ', ' + \
               'PublishTime:' + str(self.PublishTime) + ', ' + \
               'NotificationData:' + str(self.NotificationData) + ')'

    __repr__ = __str__




    __repr__ = __str__


class FindServersOnNetworkParameters(FrozenClass):
    '''
    :ivar StartingRecordId:
    :vartype StartingRecordId: UInt32
    :ivar MaxRecordsToReturn:
    :vartype MaxRecordsToReturn: UInt32
    :ivar ServerCapabilityFilter:
    :vartype ServerCapabilityFilter: String
    '''
    def __init__(self, binary=None):
        if binary is not None:
            self._binary_init(binary)
            self._freeze = True
            return
        self.StartingRecordId = 0
        self.MaxRecordsToReturn = 0
        self.ServerCapabilityFilter = []
        self._freeze = True

    def to_binary(self):
        packet = []
        packet.append(uatype_UInt32.pack(self.StartingRecordId))
        packet.append(uatype_UInt32.pack(self.MaxRecordsToReturn))
        packet.append(uatype_Int32.pack(len(self.ServerCapabilityFilter)))
        for fieldname in self.ServerCapabilityFilter:
            packet.append(pack_string(fieldname))
        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        return FindServersOnNetworkParameters(data)

    def _binary_init(self, data):
        self.StartingRecordId = uatype_UInt32.unpack(data.read(4))[0]
        self.MaxRecordsToReturn = uatype_UInt32.unpack(data.read(4))[0]
        self.ServerCapabilityFilter = unpack_uatype_array('String', data)

    def __str__(self):
        return 'FindServersOnNetworkParameters(' + 'StartingRecordId:' + str(self.StartingRecordId) + ', ' + \
               'MaxRecordsToReturn:' + str(self.MaxRecordsToReturn) + ', ' + \
               'ServerCapabilityFilter:' + str(self.ServerCapabilityFilter) + ')'

    __repr__ = __str__




    __repr__ = __str__


class NetworkGroupDataType(FrozenClass):
    '''
    :ivar ServerUri:
    :vartype ServerUri: String
    :ivar NetworkPaths:
    :vartype NetworkPaths: EndpointUrlListDataType
    '''
    def __init__(self, binary=None):
        if binary is not None:
            self._binary_init(binary)
            self._freeze = True
            return
        self.ServerUri = None
        self.NetworkPaths = []
        self._freeze = True

    def to_binary(self):
        packet = []
        packet.append(pack_string(self.ServerUri))
        packet.append(uatype_Int32.pack(len(self.NetworkPaths)))
        for fieldname in self.NetworkPaths:
            packet.append(fieldname.to_binary())
        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        return NetworkGroupDataType(data)

    def _binary_init(self, data):
        self.ServerUri = unpack_string(data)
        length = uatype_Int32.unpack(data.read(4))[0]
        array = []
        if length != -1:
            for _ in range(0, length):
                array.append(EndpointUrlListDataType.from_binary(data))
        self.NetworkPaths = array

    def __str__(self):
        return 'NetworkGroupDataType(' + 'ServerUri:' + str(self.ServerUri) + ', ' + \
               'NetworkPaths:' + str(self.NetworkPaths) + ')'

    __repr__ = __str__




    __repr__ = __str__


class QueryNextResult(FrozenClass):
    '''
    :ivar QueryDataSets:
    :vartype QueryDataSets: QueryDataSet
    :ivar RevisedContinuationPoint:
    :vartype RevisedContinuationPoint: ByteString
    '''
    def __init__(self, binary=None):
        if binary is not None:
            self._binary_init(binary)
            self._freeze = True
            return
        self.QueryDataSets = []
        self.RevisedContinuationPoint = None
        self._freeze = True

    def to_binary(self):
        packet = []
        packet.append(uatype_Int32.pack(len(self.QueryDataSets)))
        for fieldname in self.QueryDataSets:
            packet.append(fieldname.to_binary())
        packet.append(pack_bytes(self.RevisedContinuationPoint))
        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        return QueryNextResult(data)

    def _binary_init(self, data):
        length = uatype_Int32.unpack(data.read(4))[0]
        array = []
        if length != -1:
            for _ in range(0, length):
                array.append(QueryDataSet.from_binary(data))
        self.QueryDataSets = array
        self.RevisedContinuationPoint = unpack_bytes(data)

    def __str__(self):
        return 'QueryNextResult(' + 'QueryDataSets:' + str(self.QueryDataSets) + ', ' + \
               'RevisedContinuationPoint:' + str(self.RevisedContinuationPoint) + ')'

    __repr__ = __str__




    __repr__ = __str__


class FindServersOnNetworkResult(FrozenClass):
    '''
    :ivar LastCounterResetTime:
    :vartype LastCounterResetTime: DateTime
    :ivar Servers:
    :vartype Servers: ServerOnNetwork
    '''
    def __init__(self, binary=None):
        if binary is not None:
            self._binary_init(binary)
            self._freeze = True
            return
        self.LastCounterResetTime = datetime.now()
        self.Servers = []
        self._freeze = True

    def to_binary(self):
        packet = []
        packet.append(pack_datetime(self.LastCounterResetTime))
        packet.append(uatype_Int32.pack(len(self.Servers)))
        for fieldname in self.Servers:
            packet.append(fieldname.to_binary())
        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        return FindServersOnNetworkResult(data)

    def _binary_init(self, data):
        self.LastCounterResetTime = unpack_datetime(data)
        length = uatype_Int32.unpack(data.read(4))[0]
        array = []
        if length != -1:
            for _ in range(0, length):
                array.append(ServerOnNetwork.from_binary(data))
        self.Servers = array

    def __str__(self):
        return 'FindServersOnNetworkResult(' + 'LastCounterResetTime:' + str(self.LastCounterResetTime) + ', ' + \
               'Servers:' + str(self.Servers) + ')'

    __repr__ = __str__




    __repr__ = __str__


class DataChangeFilter(FrozenClass):
    '''
    :ivar Trigger:
    :vartype Trigger: DataChangeTrigger
    :ivar DeadbandType:
    :vartype DeadbandType: UInt32
    :ivar DeadbandValue:
    :vartype DeadbandValue: Double
    '''
    def __init__(self, binary=None):
        if binary is not None:
            self._binary_init(binary)
            self._freeze = True
            return
        self.Trigger = DataChangeTrigger(0)
        self.DeadbandType = 0
        self.DeadbandValue = 0
        self._freeze = True

    def to_binary(self):
        packet = []
        packet.append(uatype_UInt32.pack(self.Trigger.value))
        packet.append(uatype_UInt32.pack(self.DeadbandType))
        packet.append(uatype_Double.pack(self.DeadbandValue))
        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        return DataChangeFilter(data)

    def _binary_init(self, data):
        self.Trigger = DataChangeTrigger(uatype_UInt32.unpack(data.read(4))[0])
        self.DeadbandType = uatype_UInt32.unpack(data.read(4))[0]
        self.DeadbandValue = uatype_Double.unpack(data.read(8))[0]

    def __str__(self):
        return 'DataChangeFilter(' + 'Trigger:' + str(self.Trigger) + ', ' + \
               'DeadbandType:' + str(self.DeadbandType) + ', ' + \
               'DeadbandValue:' + str(self.DeadbandValue) + ')'

    __repr__ = __str__



		@@ 5972-6015 (lines=44) @@
5969
5970		__repr__ = __str__
5971
5972
5973		class BrowsePathResult(FrozenClass):
5974		'''
5975		The result of a translate opearation.
5976
5977		:ivar StatusCode:
5978		:vartype StatusCode: StatusCode
5979		:ivar Targets:
5980		:vartype Targets: BrowsePathTarget
5981		'''
5982		def __init__(self, binary=None):
5983		if binary is not None:
5984		self._binary_init(binary)
5985		self._freeze = True
5986		return
5987		self.StatusCode = StatusCode()
5988		self.Targets = []
5989		self._freeze = True
5990
5991		def to_binary(self):
5992		packet = []
5993		packet.append(self.StatusCode.to_binary())
5994		packet.append(uatype_Int32.pack(len(self.Targets)))
5995		for fieldname in self.Targets:
5996		packet.append(fieldname.to_binary())
5997		return b''.join(packet)
5998
5999		@staticmethod
6000		def from_binary(data):
6001		return BrowsePathResult(data)
6002
6003		def _binary_init(self, data):
6004		self.StatusCode = StatusCode.from_binary(data)
6005		length = uatype_Int32.unpack(data.read(4))[0]
6006		array = []
6007		if length != -1:
6008		for _ in range(0, length):
6009		array.append(BrowsePathTarget.from_binary(data))
6010		self.Targets = array
6011
6012		def __str__(self):
6013		return 'BrowsePathResult(' + 'StatusCode:' + str(self.StatusCode) + ', ' + \
6014		'Targets:' + str(self.Targets) + ')'
6015
6016		__repr__ = __str__
6017
6018
		@@ 10427-10469 (lines=43) @@
10424
10425		__repr__ = __str__
10426
10427
10428		class SetMonitoringModeParameters(FrozenClass):
10429		'''
10430		:ivar SubscriptionId:
10431		:vartype SubscriptionId: UInt32
10432		:ivar MonitoringMode:
10433		:vartype MonitoringMode: MonitoringMode
10434		:ivar MonitoredItemIds:
10435		:vartype MonitoredItemIds: UInt32
10436		'''
10437		def __init__(self, binary=None):
10438		if binary is not None:
10439		self._binary_init(binary)
10440		self._freeze = True
10441		return
10442		self.SubscriptionId = 0
10443		self.MonitoringMode = MonitoringMode(0)
10444		self.MonitoredItemIds = []
10445		self._freeze = True
10446
10447		def to_binary(self):
10448		packet = []
10449		packet.append(uatype_UInt32.pack(self.SubscriptionId))
10450		packet.append(uatype_UInt32.pack(self.MonitoringMode.value))
10451		packet.append(uatype_Int32.pack(len(self.MonitoredItemIds)))
10452		for fieldname in self.MonitoredItemIds:
10453		packet.append(uatype_UInt32.pack(fieldname))
10454		return b''.join(packet)
10455
10456		@staticmethod
10457		def from_binary(data):
10458		return SetMonitoringModeParameters(data)
10459
10460		def _binary_init(self, data):
10461		self.SubscriptionId = uatype_UInt32.unpack(data.read(4))[0]
10462		self.MonitoringMode = MonitoringMode(uatype_UInt32.unpack(data.read(4))[0])
10463		self.MonitoredItemIds = unpack_uatype_array('UInt32', data)
10464
10465		def __str__(self):
10466		return 'SetMonitoringModeParameters(' + 'SubscriptionId:' + str(self.SubscriptionId) + ', ' + \
10467		'MonitoringMode:' + str(self.MonitoringMode) + ', ' + \
10468		'MonitoredItemIds:' + str(self.MonitoredItemIds) + ')'
10469
10470		__repr__ = __str__
10471
10472
		@@ 11735-11776 (lines=42) @@
11732
11733		__repr__ = __str__
11734
11735
11736		class EventFieldList(FrozenClass):
11737		'''
11738		:ivar ClientHandle:
11739		:vartype ClientHandle: UInt32
11740		:ivar EventFields:
11741		:vartype EventFields: Variant
11742		'''
11743		def __init__(self, binary=None):
11744		if binary is not None:
11745		self._binary_init(binary)
11746		self._freeze = True
11747		return
11748		self.ClientHandle = 0
11749		self.EventFields = []
11750		self._freeze = True
11751
11752		def to_binary(self):
11753		packet = []
11754		packet.append(uatype_UInt32.pack(self.ClientHandle))
11755		packet.append(uatype_Int32.pack(len(self.EventFields)))
11756		for fieldname in self.EventFields:
11757		packet.append(fieldname.to_binary())
11758		return b''.join(packet)
11759
11760		@staticmethod
11761		def from_binary(data):
11762		return EventFieldList(data)
11763
11764		def _binary_init(self, data):
11765		self.ClientHandle = uatype_UInt32.unpack(data.read(4))[0]
11766		length = uatype_Int32.unpack(data.read(4))[0]
11767		array = []
11768		if length != -1:
11769		for _ in range(0, length):
11770		array.append(Variant.from_binary(data))
11771		self.EventFields = array
11772
11773		def __str__(self):
11774		return 'EventFieldList(' + 'ClientHandle:' + str(self.ClientHandle) + ', ' + \
11775		'EventFields:' + str(self.EventFields) + ')'
11776
11777		__repr__ = __str__
11778
11779
		@@ 9601-9642 (lines=42) @@
9598
9599		__repr__ = __str__
9600
9601
9602		class EventFilter(FrozenClass):
9603		'''
9604		:ivar SelectClauses:
9605		:vartype SelectClauses: SimpleAttributeOperand
9606		:ivar WhereClause:
9607		:vartype WhereClause: ContentFilter
9608		'''
9609		def __init__(self, binary=None):
9610		if binary is not None:
9611		self._binary_init(binary)
9612		self._freeze = True
9613		return
9614		self.SelectClauses = []
9615		self.WhereClause = ContentFilter()
9616		self._freeze = True
9617
9618		def to_binary(self):
9619		packet = []
9620		packet.append(uatype_Int32.pack(len(self.SelectClauses)))
9621		for fieldname in self.SelectClauses:
9622		packet.append(fieldname.to_binary())
9623		packet.append(self.WhereClause.to_binary())
9624		return b''.join(packet)
9625
9626		@staticmethod
9627		def from_binary(data):
9628		return EventFilter(data)
9629
9630		def _binary_init(self, data):
9631		length = uatype_Int32.unpack(data.read(4))[0]
9632		array = []
9633		if length != -1:
9634		for _ in range(0, length):
9635		array.append(SimpleAttributeOperand.from_binary(data))
9636		self.SelectClauses = array
9637		self.WhereClause = ContentFilter.from_binary(data)
9638
9639		def __str__(self):
9640		return 'EventFilter(' + 'SelectClauses:' + str(self.SelectClauses) + ', ' + \
9641		'WhereClause:' + str(self.WhereClause) + ')'
9642
9643		__repr__ = __str__
9644
9645
		@@ 6887-6928 (lines=42) @@
6884
6885		__repr__ = __str__
6886
6887
6888		class ContentFilterElement(FrozenClass):
6889		'''
6890		:ivar FilterOperator:
6891		:vartype FilterOperator: FilterOperator
6892		:ivar FilterOperands:
6893		:vartype FilterOperands: ExtensionObject
6894		'''
6895		def __init__(self, binary=None):
6896		if binary is not None:
6897		self._binary_init(binary)
6898		self._freeze = True
6899		return
6900		self.FilterOperator = FilterOperator(0)
6901		self.FilterOperands = []
6902		self._freeze = True
6903
6904		def to_binary(self):
6905		packet = []
6906		packet.append(uatype_UInt32.pack(self.FilterOperator.value))
6907		packet.append(uatype_Int32.pack(len(self.FilterOperands)))
6908		for fieldname in self.FilterOperands:
6909		packet.append(extensionobject_to_binary(fieldname))
6910		return b''.join(packet)
6911
6912		@staticmethod
6913		def from_binary(data):
6914		return ContentFilterElement(data)
6915
6916		def _binary_init(self, data):
6917		self.FilterOperator = FilterOperator(uatype_UInt32.unpack(data.read(4))[0])
6918		length = uatype_Int32.unpack(data.read(4))[0]
6919		array = []
6920		if length != -1:
6921		for _ in range(0, length):
6922		array.append(extensionobject_from_binary(data))
6923		self.FilterOperands = array
6924
6925		def __str__(self):
6926		return 'ContentFilterElement(' + 'FilterOperator:' + str(self.FilterOperator) + ', ' + \
6927		'FilterOperands:' + str(self.FilterOperands) + ')'
6928
6929		__repr__ = __str__
6930
6931
		@@ 2234-2275 (lines=42) @@
2231
2232		__repr__ = __str__
2233
2234
2235		class RegisterServer2Parameters(FrozenClass):
2236		'''
2237		:ivar Server:
2238		:vartype Server: RegisteredServer
2239		:ivar DiscoveryConfiguration:
2240		:vartype DiscoveryConfiguration: ExtensionObject
2241		'''
2242		def __init__(self, binary=None):
2243		if binary is not None:
2244		self._binary_init(binary)
2245		self._freeze = True
2246		return
2247		self.Server = RegisteredServer()
2248		self.DiscoveryConfiguration = []
2249		self._freeze = True
2250
2251		def to_binary(self):
2252		packet = []
2253		packet.append(self.Server.to_binary())
2254		packet.append(uatype_Int32.pack(len(self.DiscoveryConfiguration)))
2255		for fieldname in self.DiscoveryConfiguration:
2256		packet.append(extensionobject_to_binary(fieldname))
2257		return b''.join(packet)
2258
2259		@staticmethod
2260		def from_binary(data):
2261		return RegisterServer2Parameters(data)
2262
2263		def _binary_init(self, data):
2264		self.Server = RegisteredServer.from_binary(data)
2265		length = uatype_Int32.unpack(data.read(4))[0]
2266		array = []
2267		if length != -1:
2268		for _ in range(0, length):
2269		array.append(extensionobject_from_binary(data))
2270		self.DiscoveryConfiguration = array
2271
2272		def __str__(self):
2273		return 'RegisterServer2Parameters(' + 'Server:' + str(self.Server) + ', ' + \
2274		'DiscoveryConfiguration:' + str(self.DiscoveryConfiguration) + ')'
2275
2276		__repr__ = __str__
2277
2278
		@@ 11532-11579 (lines=48) @@
11529
11530		__repr__ = __str__
11531
11532
11533		class NotificationMessage(FrozenClass):
11534		'''
11535		:ivar SequenceNumber:
11536		:vartype SequenceNumber: UInt32
11537		:ivar PublishTime:
11538		:vartype PublishTime: DateTime
11539		:ivar NotificationData:
11540		:vartype NotificationData: ExtensionObject
11541		'''
11542		def __init__(self, binary=None):
11543		if binary is not None:
11544		self._binary_init(binary)
11545		self._freeze = True
11546		return
11547		self.SequenceNumber = 0
11548		self.PublishTime = datetime.now()
11549		self.NotificationData = []
11550		self._freeze = True
11551
11552		def to_binary(self):
11553		packet = []
11554		packet.append(uatype_UInt32.pack(self.SequenceNumber))
11555		packet.append(pack_datetime(self.PublishTime))
11556		packet.append(uatype_Int32.pack(len(self.NotificationData)))
11557		for fieldname in self.NotificationData:
11558		packet.append(extensionobject_to_binary(fieldname))
11559		return b''.join(packet)
11560
11561		@staticmethod
11562		def from_binary(data):
11563		return NotificationMessage(data)
11564
11565		def _binary_init(self, data):
11566		self.SequenceNumber = uatype_UInt32.unpack(data.read(4))[0]
11567		self.PublishTime = unpack_datetime(data)
11568		length = uatype_Int32.unpack(data.read(4))[0]
11569		array = []
11570		if length != -1:
11571		for _ in range(0, length):
11572		array.append(extensionobject_from_binary(data))
11573		self.NotificationData = array
11574
11575		def __str__(self):
11576		return 'NotificationMessage(' + 'SequenceNumber:' + str(self.SequenceNumber) + ', ' + \
11577		'PublishTime:' + str(self.PublishTime) + ', ' + \
11578		'NotificationData:' + str(self.NotificationData) + ')'
11579
11580		__repr__ = __str__
11581
11582
		@@ 1538-1580 (lines=43) @@
1535
1536		__repr__ = __str__
1537
1538
1539		class FindServersOnNetworkParameters(FrozenClass):
1540		'''
1541		:ivar StartingRecordId:
1542		:vartype StartingRecordId: UInt32
1543		:ivar MaxRecordsToReturn:
1544		:vartype MaxRecordsToReturn: UInt32
1545		:ivar ServerCapabilityFilter:
1546		:vartype ServerCapabilityFilter: String
1547		'''
1548		def __init__(self, binary=None):
1549		if binary is not None:
1550		self._binary_init(binary)
1551		self._freeze = True
1552		return
1553		self.StartingRecordId = 0
1554		self.MaxRecordsToReturn = 0
1555		self.ServerCapabilityFilter = []
1556		self._freeze = True
1557
1558		def to_binary(self):
1559		packet = []
1560		packet.append(uatype_UInt32.pack(self.StartingRecordId))
1561		packet.append(uatype_UInt32.pack(self.MaxRecordsToReturn))
1562		packet.append(uatype_Int32.pack(len(self.ServerCapabilityFilter)))
1563		for fieldname in self.ServerCapabilityFilter:
1564		packet.append(pack_string(fieldname))
1565		return b''.join(packet)
1566
1567		@staticmethod
1568		def from_binary(data):
1569		return FindServersOnNetworkParameters(data)
1570
1571		def _binary_init(self, data):
1572		self.StartingRecordId = uatype_UInt32.unpack(data.read(4))[0]
1573		self.MaxRecordsToReturn = uatype_UInt32.unpack(data.read(4))[0]
1574		self.ServerCapabilityFilter = unpack_uatype_array('String', data)
1575
1576		def __str__(self):
1577		return 'FindServersOnNetworkParameters(' + 'StartingRecordId:' + str(self.StartingRecordId) + ', ' + \
1578		'MaxRecordsToReturn:' + str(self.MaxRecordsToReturn) + ', ' + \
1579		'ServerCapabilityFilter:' + str(self.ServerCapabilityFilter) + ')'
1580
1581		__repr__ = __str__
1582
1583
		@@ 12706-12747 (lines=42) @@
12703
12704		__repr__ = __str__
12705
12706
12707		class NetworkGroupDataType(FrozenClass):
12708		'''
12709		:ivar ServerUri:
12710		:vartype ServerUri: String
12711		:ivar NetworkPaths:
12712		:vartype NetworkPaths: EndpointUrlListDataType
12713		'''
12714		def __init__(self, binary=None):
12715		if binary is not None:
12716		self._binary_init(binary)
12717		self._freeze = True
12718		return
12719		self.ServerUri = None
12720		self.NetworkPaths = []
12721		self._freeze = True
12722
12723		def to_binary(self):
12724		packet = []
12725		packet.append(pack_string(self.ServerUri))
12726		packet.append(uatype_Int32.pack(len(self.NetworkPaths)))
12727		for fieldname in self.NetworkPaths:
12728		packet.append(fieldname.to_binary())
12729		return b''.join(packet)
12730
12731		@staticmethod
12732		def from_binary(data):
12733		return NetworkGroupDataType(data)
12734
12735		def _binary_init(self, data):
12736		self.ServerUri = unpack_string(data)
12737		length = uatype_Int32.unpack(data.read(4))[0]
12738		array = []
12739		if length != -1:
12740		for _ in range(0, length):
12741		array.append(EndpointUrlListDataType.from_binary(data))
12742		self.NetworkPaths = array
12743
12744		def __str__(self):
12745		return 'NetworkGroupDataType(' + 'ServerUri:' + str(self.ServerUri) + ', ' + \
12746		'NetworkPaths:' + str(self.NetworkPaths) + ')'
12747
12748		__repr__ = __str__
12749
12750
		@@ 7611-7652 (lines=42) @@
7608
7609		__repr__ = __str__
7610
7611
7612		class QueryNextResult(FrozenClass):
7613		'''
7614		:ivar QueryDataSets:
7615		:vartype QueryDataSets: QueryDataSet
7616		:ivar RevisedContinuationPoint:
7617		:vartype RevisedContinuationPoint: ByteString
7618		'''
7619		def __init__(self, binary=None):
7620		if binary is not None:
7621		self._binary_init(binary)
7622		self._freeze = True
7623		return
7624		self.QueryDataSets = []
7625		self.RevisedContinuationPoint = None
7626		self._freeze = True
7627
7628		def to_binary(self):
7629		packet = []
7630		packet.append(uatype_Int32.pack(len(self.QueryDataSets)))
7631		for fieldname in self.QueryDataSets:
7632		packet.append(fieldname.to_binary())
7633		packet.append(pack_bytes(self.RevisedContinuationPoint))
7634		return b''.join(packet)
7635
7636		@staticmethod
7637		def from_binary(data):
7638		return QueryNextResult(data)
7639
7640		def _binary_init(self, data):
7641		length = uatype_Int32.unpack(data.read(4))[0]
7642		array = []
7643		if length != -1:
7644		for _ in range(0, length):
7645		array.append(QueryDataSet.from_binary(data))
7646		self.QueryDataSets = array
7647		self.RevisedContinuationPoint = unpack_bytes(data)
7648
7649		def __str__(self):
7650		return 'QueryNextResult(' + 'QueryDataSets:' + str(self.QueryDataSets) + ', ' + \
7651		'RevisedContinuationPoint:' + str(self.RevisedContinuationPoint) + ')'
7652
7653		__repr__ = __str__
7654
7655
		@@ 1626-1667 (lines=42) @@
1623
1624		__repr__ = __str__
1625
1626
1627		class FindServersOnNetworkResult(FrozenClass):
1628		'''
1629		:ivar LastCounterResetTime:
1630		:vartype LastCounterResetTime: DateTime
1631		:ivar Servers:
1632		:vartype Servers: ServerOnNetwork
1633		'''
1634		def __init__(self, binary=None):
1635		if binary is not None:
1636		self._binary_init(binary)
1637		self._freeze = True
1638		return
1639		self.LastCounterResetTime = datetime.now()
1640		self.Servers = []
1641		self._freeze = True
1642
1643		def to_binary(self):
1644		packet = []
1645		packet.append(pack_datetime(self.LastCounterResetTime))
1646		packet.append(uatype_Int32.pack(len(self.Servers)))
1647		for fieldname in self.Servers:
1648		packet.append(fieldname.to_binary())
1649		return b''.join(packet)
1650
1651		@staticmethod
1652		def from_binary(data):
1653		return FindServersOnNetworkResult(data)
1654
1655		def _binary_init(self, data):
1656		self.LastCounterResetTime = unpack_datetime(data)
1657		length = uatype_Int32.unpack(data.read(4))[0]
1658		array = []
1659		if length != -1:
1660		for _ in range(0, length):
1661		array.append(ServerOnNetwork.from_binary(data))
1662		self.Servers = array
1663
1664		def __str__(self):
1665		return 'FindServersOnNetworkResult(' + 'LastCounterResetTime:' + str(self.LastCounterResetTime) + ', ' + \
1666		'Servers:' + str(self.Servers) + ')'
1667
1668		__repr__ = __str__
1669
1670
		@@ 9558-9598 (lines=41) @@
9555
9556		__repr__ = __str__
9557
9558
9559		class DataChangeFilter(FrozenClass):
9560		'''
9561		:ivar Trigger:
9562		:vartype Trigger: DataChangeTrigger
9563		:ivar DeadbandType:
9564		:vartype DeadbandType: UInt32
9565		:ivar DeadbandValue:
9566		:vartype DeadbandValue: Double
9567		'''
9568		def __init__(self, binary=None):
9569		if binary is not None:
9570		self._binary_init(binary)
9571		self._freeze = True
9572		return
9573		self.Trigger = DataChangeTrigger(0)
9574		self.DeadbandType = 0
9575		self.DeadbandValue = 0
9576		self._freeze = True
9577
9578		def to_binary(self):
9579		packet = []
9580		packet.append(uatype_UInt32.pack(self.Trigger.value))
9581		packet.append(uatype_UInt32.pack(self.DeadbandType))
9582		packet.append(uatype_Double.pack(self.DeadbandValue))
9583		return b''.join(packet)
9584
9585		@staticmethod
9586		def from_binary(data):
9587		return DataChangeFilter(data)
9588
9589		def _binary_init(self, data):
9590		self.Trigger = DataChangeTrigger(uatype_UInt32.unpack(data.read(4))[0])
9591		self.DeadbandType = uatype_UInt32.unpack(data.read(4))[0]
9592		self.DeadbandValue = uatype_Double.unpack(data.read(8))[0]
9593
9594		def __str__(self):
9595		return 'DataChangeFilter(' + 'Trigger:' + str(self.Trigger) + ', ' + \
9596		'DeadbandType:' + str(self.DeadbandType) + ', ' + \
9597		'DeadbandValue:' + str(self.DeadbandValue) + ')'
9598
9599		__repr__ = __str__
9600
9601

FreeOpcUa / python-opcua

Code Duplication Length = 41-48 lines in 12 locations

opcua/ua/uaprotocol_auto.py 12 locations