Code Duplication    Length = 55-59 lines in 6 locations

opcua/ua/uaprotocol_auto.py 6 locations

@@ 12070-12128 (lines=59) @@
12067
        'TypeId': 'NodeId',
12068
        'ResponseHeader': 'ResponseHeader',
12069
        'Parameters': 'SetMonitoringModeResult',
12070
               }
12071
12072
    def __init__(self, binary=None):
12073
        if binary is not None:
12074
            self._binary_init(binary)
12075
            self._freeze = True
12076
            return
12077
        self.TypeId = FourByteNodeId(ObjectIds.SetMonitoringModeResponse_Encoding_DefaultBinary)
12078
        self.ResponseHeader = ResponseHeader()
12079
        self.Parameters = SetMonitoringModeResult()
12080
        self._freeze = True
12081
12082
    def to_binary(self):
12083
        packet = []
12084
        packet.append(self.TypeId.to_binary())
12085
        packet.append(self.ResponseHeader.to_binary())
12086
        packet.append(self.Parameters.to_binary())
12087
        return b''.join(packet)
12088
12089
    @staticmethod
12090
    def from_binary(data):
12091
        return SetMonitoringModeResponse(data)
12092
12093
    def _binary_init(self, data):
12094
        self.TypeId = NodeId.from_binary(data)
12095
        self.ResponseHeader = ResponseHeader.from_binary(data)
12096
        self.Parameters = SetMonitoringModeResult.from_binary(data)
12097
12098
    def __str__(self):
12099
        return 'SetMonitoringModeResponse(' + 'TypeId:' + str(self.TypeId) + ', ' + \
12100
               'ResponseHeader:' + str(self.ResponseHeader) + ', ' + \
12101
               'Parameters:' + str(self.Parameters) + ')'
12102
12103
    __repr__ = __str__
12104
12105
12106
class SetTriggeringParameters(FrozenClass):
12107
    '''
12108
    :ivar SubscriptionId:
12109
    :vartype SubscriptionId: UInt32
12110
    :ivar TriggeringItemId:
12111
    :vartype TriggeringItemId: UInt32
12112
    :ivar LinksToAdd:
12113
    :vartype LinksToAdd: UInt32
12114
    :ivar LinksToRemove:
12115
    :vartype LinksToRemove: UInt32
12116
    '''
12117
12118
    ua_types = {
12119
        'SubscriptionId': 'UInt32',
12120
        'TriggeringItemId': 'UInt32',
12121
        'LinksToAdd': 'UInt32',
12122
        'LinksToRemove': 'UInt32',
12123
               }
12124
12125
    def __init__(self, binary=None):
12126
        if binary is not None:
12127
            self._binary_init(binary)
12128
            self._freeze = True
12129
            return
12130
        self.SubscriptionId = 0
12131
        self.TriggeringItemId = 0
@@ 1600-1656 (lines=57) @@
1597
    __repr__ = __str__
1598
1599
1600
class ServerOnNetwork(FrozenClass):
1601
    '''
1602
    :ivar RecordId:
1603
    :vartype RecordId: UInt32
1604
    :ivar ServerName:
1605
    :vartype ServerName: String
1606
    :ivar DiscoveryUrl:
1607
    :vartype DiscoveryUrl: String
1608
    :ivar ServerCapabilities:
1609
    :vartype ServerCapabilities: String
1610
    '''
1611
1612
    ua_types = {
1613
        'RecordId': 'UInt32',
1614
        'ServerName': 'String',
1615
        'DiscoveryUrl': 'String',
1616
        'ServerCapabilities': 'String',
1617
               }
1618
1619
    def __init__(self, binary=None):
1620
        if binary is not None:
1621
            self._binary_init(binary)
1622
            self._freeze = True
1623
            return
1624
        self.RecordId = 0
1625
        self.ServerName = None
1626
        self.DiscoveryUrl = None
1627
        self.ServerCapabilities = []
1628
        self._freeze = True
1629
1630
    def to_binary(self):
1631
        packet = []
1632
        packet.append(uabin.Primitives.UInt32.pack(self.RecordId))
1633
        packet.append(uabin.Primitives.String.pack(self.ServerName))
1634
        packet.append(uabin.Primitives.String.pack(self.DiscoveryUrl))
1635
        packet.append(uabin.Primitives.Int32.pack(len(self.ServerCapabilities)))
1636
        for fieldname in self.ServerCapabilities:
1637
            packet.append(uabin.Primitives.String.pack(fieldname))
1638
        return b''.join(packet)
1639
1640
    @staticmethod
1641
    def from_binary(data):
1642
        return ServerOnNetwork(data)
1643
1644
    def _binary_init(self, data):
1645
        self.RecordId = uabin.Primitives.UInt32.unpack(data)
1646
        self.ServerName = uabin.Primitives.String.unpack(data)
1647
        self.DiscoveryUrl = uabin.Primitives.String.unpack(data)
1648
        self.ServerCapabilities = uabin.Primitives.String.unpack_array(data)
1649
1650
    def __str__(self):
1651
        return 'ServerOnNetwork(' + 'RecordId:' + str(self.RecordId) + ', ' + \
1652
               'ServerName:' + str(self.ServerName) + ', ' + \
1653
               'DiscoveryUrl:' + str(self.DiscoveryUrl) + ', ' + \
1654
               'ServerCapabilities:' + str(self.ServerCapabilities) + ')'
1655
1656
    __repr__ = __str__
1657
1658
1659
class FindServersOnNetworkParameters(FrozenClass):
@@ 3458-3514 (lines=57) @@
3455
    __repr__ = __str__
3456
3457
3458
class UserNameIdentityToken(FrozenClass):
3459
    '''
3460
    A token representing a user identified by a user name and password.
3461
3462
    :ivar PolicyId:
3463
    :vartype PolicyId: String
3464
    :ivar UserName:
3465
    :vartype UserName: String
3466
    :ivar Password:
3467
    :vartype Password: ByteString
3468
    :ivar EncryptionAlgorithm:
3469
    :vartype EncryptionAlgorithm: String
3470
    '''
3471
3472
    ua_types = {
3473
        'PolicyId': 'String',
3474
        'UserName': 'String',
3475
        'Password': 'ByteString',
3476
        'EncryptionAlgorithm': 'String',
3477
               }
3478
3479
    def __init__(self, binary=None):
3480
        if binary is not None:
3481
            self._binary_init(binary)
3482
            self._freeze = True
3483
            return
3484
        self.PolicyId = None
3485
        self.UserName = None
3486
        self.Password = None
3487
        self.EncryptionAlgorithm = None
3488
        self._freeze = True
3489
3490
    def to_binary(self):
3491
        packet = []
3492
        packet.append(uabin.Primitives.String.pack(self.PolicyId))
3493
        packet.append(uabin.Primitives.String.pack(self.UserName))
3494
        packet.append(uabin.Primitives.ByteString.pack(self.Password))
3495
        packet.append(uabin.Primitives.String.pack(self.EncryptionAlgorithm))
3496
        return b''.join(packet)
3497
3498
    @staticmethod
3499
    def from_binary(data):
3500
        return UserNameIdentityToken(data)
3501
3502
    def _binary_init(self, data):
3503
        self.PolicyId = uabin.Primitives.String.unpack(data)
3504
        self.UserName = uabin.Primitives.String.unpack(data)
3505
        self.Password = uabin.Primitives.ByteString.unpack(data)
3506
        self.EncryptionAlgorithm = uabin.Primitives.String.unpack(data)
3507
3508
    def __str__(self):
3509
        return 'UserNameIdentityToken(' + 'PolicyId:' + str(self.PolicyId) + ', ' + \
3510
               'UserName:' + str(self.UserName) + ', ' + \
3511
               'Password:' + str(self.Password) + ', ' + \
3512
               'EncryptionAlgorithm:' + str(self.EncryptionAlgorithm) + ')'
3513
3514
    __repr__ = __str__
3515
3516
3517
class X509IdentityToken(FrozenClass):
@@ 14531-14585 (lines=55) @@
14528
               }
14529
14530
    def __init__(self, binary=None):
14531
        if binary is not None:
14532
            self._binary_init(binary)
14533
            self._freeze = True
14534
            return
14535
        self.ServerUri = None
14536
        self.NetworkPaths = []
14537
        self._freeze = True
14538
14539
    def to_binary(self):
14540
        packet = []
14541
        packet.append(uabin.Primitives.String.pack(self.ServerUri))
14542
        packet.append(uabin.Primitives.Int32.pack(len(self.NetworkPaths)))
14543
        for fieldname in self.NetworkPaths:
14544
            packet.append(fieldname.to_binary())
14545
        return b''.join(packet)
14546
14547
    @staticmethod
14548
    def from_binary(data):
14549
        return NetworkGroupDataType(data)
14550
14551
    def _binary_init(self, data):
14552
        self.ServerUri = uabin.Primitives.String.unpack(data)
14553
        length = uabin.Primitives.Int32.unpack(data)
14554
        array = []
14555
        if length != -1:
14556
            for _ in range(0, length):
14557
                array.append(EndpointUrlListDataType.from_binary(data))
14558
        self.NetworkPaths = array
14559
14560
    def __str__(self):
14561
        return 'NetworkGroupDataType(' + 'ServerUri:' + str(self.ServerUri) + ', ' + \
14562
               'NetworkPaths:' + str(self.NetworkPaths) + ')'
14563
14564
    __repr__ = __str__
14565
14566
14567
class SamplingIntervalDiagnosticsDataType(FrozenClass):
14568
    '''
14569
    :ivar SamplingInterval:
14570
    :vartype SamplingInterval: Double
14571
    :ivar MonitoredItemCount:
14572
    :vartype MonitoredItemCount: UInt32
14573
    :ivar MaxMonitoredItemCount:
14574
    :vartype MaxMonitoredItemCount: UInt32
14575
    :ivar DisabledMonitoredItemCount:
14576
    :vartype DisabledMonitoredItemCount: UInt32
14577
    '''
14578
14579
    ua_types = {
14580
        'SamplingInterval': 'Double',
14581
        'MonitoredItemCount': 'UInt32',
14582
        'MaxMonitoredItemCount': 'UInt32',
14583
        'DisabledMonitoredItemCount': 'UInt32',
14584
               }
14585
14586
    def __init__(self, binary=None):
14587
        if binary is not None:
14588
            self._binary_init(binary)
@@ 12603-12657 (lines=55) @@
12600
        'TypeId': 'NodeId',
12601
        'RequestHeader': 'RequestHeader',
12602
        'Parameters': 'CreateSubscriptionParameters',
12603
               }
12604
12605
    def __init__(self, binary=None):
12606
        if binary is not None:
12607
            self._binary_init(binary)
12608
            self._freeze = True
12609
            return
12610
        self.TypeId = FourByteNodeId(ObjectIds.CreateSubscriptionRequest_Encoding_DefaultBinary)
12611
        self.RequestHeader = RequestHeader()
12612
        self.Parameters = CreateSubscriptionParameters()
12613
        self._freeze = True
12614
12615
    def to_binary(self):
12616
        packet = []
12617
        packet.append(self.TypeId.to_binary())
12618
        packet.append(self.RequestHeader.to_binary())
12619
        packet.append(self.Parameters.to_binary())
12620
        return b''.join(packet)
12621
12622
    @staticmethod
12623
    def from_binary(data):
12624
        return CreateSubscriptionRequest(data)
12625
12626
    def _binary_init(self, data):
12627
        self.TypeId = NodeId.from_binary(data)
12628
        self.RequestHeader = RequestHeader.from_binary(data)
12629
        self.Parameters = CreateSubscriptionParameters.from_binary(data)
12630
12631
    def __str__(self):
12632
        return 'CreateSubscriptionRequest(' + 'TypeId:' + str(self.TypeId) + ', ' + \
12633
               'RequestHeader:' + str(self.RequestHeader) + ', ' + \
12634
               'Parameters:' + str(self.Parameters) + ')'
12635
12636
    __repr__ = __str__
12637
12638
12639
class CreateSubscriptionResult(FrozenClass):
12640
    '''
12641
    :ivar SubscriptionId:
12642
    :vartype SubscriptionId: UInt32
12643
    :ivar RevisedPublishingInterval:
12644
    :vartype RevisedPublishingInterval: Double
12645
    :ivar RevisedLifetimeCount:
12646
    :vartype RevisedLifetimeCount: UInt32
12647
    :ivar RevisedMaxKeepAliveCount:
12648
    :vartype RevisedMaxKeepAliveCount: UInt32
12649
    '''
12650
12651
    ua_types = {
12652
        'SubscriptionId': 'UInt32',
12653
        'RevisedPublishingInterval': 'Double',
12654
        'RevisedLifetimeCount': 'UInt32',
12655
        'RevisedMaxKeepAliveCount': 'UInt32',
12656
               }
12657
12658
    def __init__(self, binary=None):
12659
        if binary is not None:
12660
            self._binary_init(binary)
@@ 2630-2686 (lines=57) @@
2627
    __repr__ = __str__
2628
2629
2630
class ChannelSecurityToken(FrozenClass):
2631
    '''
2632
    The token that identifies a set of keys for an active secure channel.
2633
2634
    :ivar ChannelId:
2635
    :vartype ChannelId: UInt32
2636
    :ivar TokenId:
2637
    :vartype TokenId: UInt32
2638
    :ivar CreatedAt:
2639
    :vartype CreatedAt: DateTime
2640
    :ivar RevisedLifetime:
2641
    :vartype RevisedLifetime: UInt32
2642
    '''
2643
2644
    ua_types = {
2645
        'ChannelId': 'UInt32',
2646
        'TokenId': 'UInt32',
2647
        'CreatedAt': 'DateTime',
2648
        'RevisedLifetime': 'UInt32',
2649
               }
2650
2651
    def __init__(self, binary=None):
2652
        if binary is not None:
2653
            self._binary_init(binary)
2654
            self._freeze = True
2655
            return
2656
        self.ChannelId = 0
2657
        self.TokenId = 0
2658
        self.CreatedAt = datetime.utcnow()
2659
        self.RevisedLifetime = 0
2660
        self._freeze = True
2661
2662
    def to_binary(self):
2663
        packet = []
2664
        packet.append(uabin.Primitives.UInt32.pack(self.ChannelId))
2665
        packet.append(uabin.Primitives.UInt32.pack(self.TokenId))
2666
        packet.append(uabin.Primitives.DateTime.pack(self.CreatedAt))
2667
        packet.append(uabin.Primitives.UInt32.pack(self.RevisedLifetime))
2668
        return b''.join(packet)
2669
2670
    @staticmethod
2671
    def from_binary(data):
2672
        return ChannelSecurityToken(data)
2673
2674
    def _binary_init(self, data):
2675
        self.ChannelId = uabin.Primitives.UInt32.unpack(data)
2676
        self.TokenId = uabin.Primitives.UInt32.unpack(data)
2677
        self.CreatedAt = uabin.Primitives.DateTime.unpack(data)
2678
        self.RevisedLifetime = uabin.Primitives.UInt32.unpack(data)
2679
2680
    def __str__(self):
2681
        return 'ChannelSecurityToken(' + 'ChannelId:' + str(self.ChannelId) + ', ' + \
2682
               'TokenId:' + str(self.TokenId) + ', ' + \
2683
               'CreatedAt:' + str(self.CreatedAt) + ', ' + \
2684
               'RevisedLifetime:' + str(self.RevisedLifetime) + ')'
2685
2686
    __repr__ = __str__
2687
2688
2689
class OpenSecureChannelParameters(FrozenClass):