opcua.uadiscover() - Code Metrics - Inspection of "add support for find_servers_on_network to client..." - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — dev ( a35960...17cd27 )

by Olivier

created 2015-12-02 20:53 UTC

opcua.uadiscover() C

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
dl	0
loc	43
rs	5.5
cc	7

import logging
import sys
import argparse
from datetime import datetime
import code
from enum import Enum

from opcua import ua
from opcua import Client
from opcua import Node


def add_minimum_args(parser):
    parser.add_argument("-u",
                        "--url",
                        help="URL of OPC UA server (for example: opc.tcp://example.org:4840)",
                        default='opc.tcp://localhost:4841',
                        metavar="URL")
    parser.add_argument("-v",
                        "--verbose",
                        dest="loglevel",
                        choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'],
                        default='WARNING',
                        help="Set log level")
    parser.add_argument("--timeout",
                        dest="timeout",
                        type=int,
                        default=1,
                        help="Set socket timeout (NOT the diverse UA timeouts)")


def add_common_args(parser):
    add_minimum_args(parser)
    parser.add_argument("-n",
                        "--nodeid",
                        help="Fully-qualified node ID (for example: i=85). Default: root node",
                        default='i=84',
                        metavar="NODE")
    parser.add_argument("-p",
                        "--path",
                        help="Comma separated browse path to the node starting at nodeid (for example: 3:Mybject,3:MyVariable)",
                        default='',
                        metavar="BROWSEPATH")
    parser.add_argument("-i",
                        "--namespace",
                        help="Default namespace",
                        type=int,
                        default=0,
                        metavar="NAMESPACE")


def get_node(client, args):
    node = client.get_node(args.nodeid)
    if args.path:
        node = node.get_child(args.path.split(","))


def uaread():
    parser = argparse.ArgumentParser(description="Read attribute of a node, per default reads value of a node")
    add_common_args(parser)
    parser.add_argument("-a",
                        "--attribute",
                        dest="attribute",
                        type=int,
                        default=ua.AttributeIds.Value,
                        help="Set attribute to read")
    parser.add_argument("-t",
                        "--datatype",
                        dest="datatype",
                        default="python",
                        choices=['python', 'variant', 'datavalue'],
                        help="Data type to return")

    args = parser.parse_args()
    if args.nodeid == "i=84" and args.path == "" and args.attribute == ua.AttributeIds.Value:
        parser.print_usage()
        print("uaread: error: A NodeId or BrowsePath is required")
        sys.exit(1)
    logging.basicConfig(format="%(levelname)s: %(message)s", level=getattr(logging, args.loglevel))

    client = Client(args.url, timeout=args.timeout)
    client.connect()
    try:
        node = client.get_node(args.nodeid)
        if args.path:
            node = node.get_child(args.path.split(","))
        attr = node.get_attribute(args.attribute)
        if args.datatype == "python":
            print(attr.Value.Value)
        elif args.datatype == "variant":
            print(attr.Value)
        else:
            print(attr)
    finally:
        client.disconnect()
    sys.exit(0)
    print(args)


def _args_to_array(val, array):
    if array == "guess":
        if "," in val:
            array = "true"
    if array == "true":
        val = val.split(",")
    return val


def _arg_to_bool(val):
    if val in ("true", "True"):
        return True
    else:
        return False


def _arg_to_variant(val, array, ptype, varianttype=None):
    val = _args_to_array(val, array)
    if isinstance(val, list):
        val = [ptype(i) for i in val]
    else:
        val = ptype(val)
    if varianttype:
        return ua.Variant(val, varianttype)
    else:
        return ua.Variant(val)


def _val_to_variant(val, args):
    array = args.array
    if args.datatype == "guess":
        if val in ("true", "True", "false", "False"):
            return _arg_to_variant(val, array, _arg_to_bool)
        # FIXME: guess bool value
        try:
            return _arg_to_variant(val, array, int)
        except ValueError:
            try:
                return _arg_to_variant(val, array, float)
            except ValueError:
                return _arg_to_variant(val, array, str)
    elif args.datatype == "bool":
        if val in ("1", "True", "true"):
            return ua.Variant(True, ua.VariantType.Boolean)
        else:
            return ua.Variant(False, ua.VariantType.Boolean)
    elif args.datatype == "sbyte":
        return _arg_to_variant(val, array, int, ua.VariantType.SByte)
    elif args.datatype == "byte":
        return _arg_to_variant(val, array, int, ua.VariantType.Byte)
    #elif args.datatype == "uint8":
        #return _arg_to_variant(val, array, int, ua.VariantType.Byte)
    elif args.datatype == "uint16":
        return _arg_to_variant(val, array, int, ua.VariantType.UInt16)
    elif args.datatype == "uint32":
        return _arg_to_variant(val, array, int, ua.VariantType.UInt32)
    elif args.datatype == "uint64":
        return _arg_to_variant(val, array, int, ua.VariantType.UInt64)
    #elif args.datatype == "int8":
        #return ua.Variant(int(val), ua.VariantType.Int8)
    elif args.datatype == "int16":
        return _arg_to_variant(val, array, int, ua.VariantType.Int16)
    elif args.datatype == "int32":
        return _arg_to_variant(val, array, int, ua.VariantType.Int32)
    elif args.datatype == "int64":
        return _arg_to_variant(val, array, int, ua.VariantType.Int64)
    elif args.datatype == "float":
        return _arg_to_variant(val, array, float, ua.VariantType.Float)
    elif args.datatype == "double":
        return _arg_to_variant(val, array, float, ua.VariantType.Double)
    elif args.datatype == "string":
        return _arg_to_variant(val, array, str, ua.VariantType.String)
    elif args.datatype == "datetime":
        raise NotImplementedError
    elif args.datatype == "Guid":
        return _arg_to_variant(val, array, bytes, ua.VariantType.Guid)
    elif args.datatype == "ByteString":
        return _arg_to_variant(val, array, bytes, ua.VariantType.ByteString)
    elif args.datatype == "xml":
        return _arg_to_variant(val, array, str, ua.VariantType.XmlElement)
    elif args.datatype == "nodeid":
        return _arg_to_variant(val, array, ua.NodeId.from_string, ua.VariantType.NodeId)
    elif args.datatype == "expandednodeid":
        return _arg_to_variant(val, array, ua.ExpandedNodeId.from_string, ua.VariantType.ExpandedNodeId)
    elif args.datatype == "statuscode":
        return _arg_to_variant(val, array, int, ua.VariantType.StatusCode)
    elif args.datatype in ("qualifiedname", "browsename"):
        return _arg_to_variant(val, array, ua.QualifiedName.from_string, ua.VariantType.QualifiedName)
    elif args.datatype == "LocalizedText":
        return _arg_to_variant(val, array, ua.LocalizedText, ua.VariantTypeLocalizedText)


def uawrite():
    parser = argparse.ArgumentParser(description="Write attribute of a node, per default write value of node")
    add_common_args(parser)
    parser.add_argument("-a",
                        "--attribute",
                        dest="attribute",
                        type=int,
                        default=ua.AttributeIds.Value,
                        help="Set attribute to read")
    parser.add_argument("-l",
                        "--list",
                        "--array",
                        dest="array",
                        default="guess",
                        choices=["guess", "true", "false"],
                        help="Value is an array")
    parser.add_argument("-t",
                        "--datatype",
                        dest="datatype",
                        default="guess",
                        choices=["guess", 'byte', 'sbyte', 'nodeid', 'expandednodeid', 'qualifiedname', 'browsename', 'string', 'float', 'double', 'int16', 'int32', "int64", 'uint16', 'uint32', 'uint64', "bool", "string", 'datetime', 'bytestring', 'xmlelement', 'statuscode', 'localizedtext'],  
                        help="Data type to return")
    parser.add_argument("value",
                        help="Value to be written",
                        metavar="VALUE")
    args = parser.parse_args()
    if args.nodeid == "i=84" and args.path == "" and args.attribute == ua.AttributeIds.Value:
        parser.print_usage()
        print("uaread: error: A NodeId or BrowsePath is required")
        sys.exit(1)
    logging.basicConfig(format="%(levelname)s: %(message)s", level=getattr(logging, args.loglevel))

    client = Client(args.url, timeout=args.timeout)
    client.connect()
    try:
        node = client.get_node(args.nodeid)
        if args.path:
            node = node.get_child(args.path.split(","))
        val = _val_to_variant(args.value, args)
        node.set_attribute(args.attribute, ua.DataValue(val))
    finally:
        client.disconnect()
    sys.exit(0)
    print(args)


def uals():
    parser = argparse.ArgumentParser(description="Browse OPC-UA node and print result")
    add_common_args(parser)
    parser.add_argument("-l",
                        dest="long_format",
                        const=3,
                        nargs="?",
                        type=int,
                        help="use a long listing format")
    parser.add_argument("-d",
                        "--depth",
                        default=1,
                        type=int,
                        help="Browse depth")

    args = parser.parse_args()
    if args.long_format is None:
        args.long_format = 1
    logging.basicConfig(format="%(levelname)s: %(message)s", level=getattr(logging, args.loglevel))

    client = Client(args.url, timeout=args.timeout)
    client.connect()
    try:
        node = client.get_node(args.nodeid)
        if args.path:
            node = node.get_child(args.path.split(","))
        print("Browsing node {} at {}\n".format(node, args.url))
        if args.long_format == 0:
            _lsprint_0(node, args.depth - 1)
        elif args.long_format == 1:
            _lsprint_1(node, args.depth - 1)
        else:
            _lsprint_long(node, args.depth - 1)
    finally:
        client.disconnect()
    sys.exit(0)
    print(args)


def _lsprint_0(node, depth, indent=""):
    if not indent:
        print("{:30} {:25}".format("DisplayName", "NodeId"))
        print("")
    for desc in node.get_children_descriptions():
        print("{}{:30} {:25}".format(indent, desc.DisplayName.to_string(), desc.NodeId.to_string()))
        if depth:
            _lsprint_0(Node(node.server, desc.NodeId), depth - 1, indent + "  ")


def _lsprint_1(node, depth, indent=""):
    if not indent:
        print("{:30} {:25} {:25} {:25}".format("DisplayName", "NodeId", "BrowseName", "Value"))
        print("")

    for desc in node.get_children_descriptions():
        if desc.NodeClass == ua.NodeClass.Variable:
            val = Node(node.server, desc.NodeId).get_value()
            print("{}{:30} {!s:25} {!s:25}, {!s:3}".format(indent, desc.DisplayName.to_string(), desc.NodeId.to_string(), desc.BrowseName.to_string(), val))
        else:
            print("{}{:30} {!s:25} {!s:25}".format(indent, desc.DisplayName.to_string(), desc.NodeId.to_string(), desc.BrowseName.to_string()))
        if depth:
            _lsprint_1(Node(node.server, desc.NodeId), depth - 1, indent + "  ")


def _lsprint_long(pnode, depth, indent=""):
    if not indent:
        print("{:30} {:25} {:25} {:10} {:30} {:25}".format("DisplayName", "NodeId", "BrowseName", "DataType", "Timestamp", "Value"))
        print("")
    for node in pnode.get_children():
        attrs = node.get_attributes([ua.AttributeIds.DisplayName, 
                                     ua.AttributeIds.BrowseName,
                                     ua.AttributeIds.NodeClass,
                                     ua.AttributeIds.WriteMask,
                                     ua.AttributeIds.UserWriteMask,
                                     ua.AttributeIds.DataType,
                                     ua.AttributeIds.Value])
        name, bname, nclass, mask, umask, dtype, val = [attr.Value.Value for attr in attrs]
        update = attrs[-1].ServerTimestamp
        if nclass == ua.NodeClass.Variable:
            print("{}{:30} {:25} {:25} {:10} {!s:30} {!s:25}".format(indent, name.to_string(), node.nodeid.to_string(), bname.to_string(), dtype.to_string(), update, val))
        else:
            print("{}{:30} {:25} {:25}".format(indent, name.to_string(), bname.to_string(), node.nodeid.to_string()))
        if depth:
            _lsprint_long(node, depth - 1, indent + "  ")


class SubHandler(object):

    def data_change(self, handle, node, val, attr):
        print("New data change event", handle, node, val, attr)

    def event(self, handle, event):
        print("New event", handle, event)


def uasubscribe():
    parser = argparse.ArgumentParser(description="Subscribe to a node and print results")
    add_common_args(parser)
    parser.add_argument("-t",
                        "--eventtype",
                        dest="eventtype",
                        default="datachange",
                        choices=['datachange', 'event'],
                        help="Event type to subscribe to")

    args = parser.parse_args()
    if args.nodeid == "i=84" and args.path == "":
        parser.print_usage()
        print("uaread: error: The NodeId or BrowsePath of a variable is required")
        sys.exit(1)
    logging.basicConfig(format="%(levelname)s: %(message)s", level=getattr(logging, args.loglevel))

    client = Client(args.url, timeout=args.timeout)
    client.connect()
    try:
        node = client.get_node(args.nodeid)
        if args.path:
            node = node.get_child(args.path.split(","))
        handler = SubHandler()
        sub = client.create_subscription(500, handler)
        if args.eventtype == "datachange":
            sub.subscribe_data_change(node)
        else:
            sub.subscribe_events(node)
        glbs = globals()
        glbs.update(locals())
        shell = code.InteractiveConsole(vars)
        shell.interact()
    finally:
        client.disconnect()
    sys.exit(0)
    print(args)


# converts numeric value to its enum name.
def enum_to_string(klass, value):
    if isinstance(value, Enum):
        return value.name
    # if value is not a subtype of Enum, try to find a constant
    # with this value in this class
    for k, v in vars(klass).items():
        if not k.startswith('__') and v == value:
            return k
    return 'Unknown {} ({})'.format(klass.__name__, value)


def application_to_strings(app):
    result = []
    result.append(('Application URI', app.ApplicationUri))
    optionals = [
        ('Product URI', app.ProductUri),
        ('Application Name', app.ApplicationName.to_string()),
        ('Application Type', enum_to_string(ua.ApplicationType, app.ApplicationType)),
        ('Gateway Server URI', app.GatewayServerUri),
        ('Discovery Profile URI', app.DiscoveryProfileUri),
    ]
    for (n, v) in optionals:
        if v:
            result.append((n, v))
    for url in app.DiscoveryUrls:
        result.append(('Discovery URL', url))
    return result  # ['{}: {}'.format(n, v) for (n, v) in result]


def endpoint_to_strings(ep):
    result = [('Endpoint URL', ep.EndpointUrl)]
    result += application_to_strings(ep.Server)
    result += [
        ('Server Certificate', len(ep.ServerCertificate)),
        ('Security Mode', enum_to_string(ua.MessageSecurityMode, ep.SecurityMode)),
        ('Security Policy URI', ep.SecurityPolicyUri)]
    for tok in ep.UserIdentityTokens:
        result += [
            ('User policy', tok.PolicyId),
            ('  Token type', enum_to_string(ua.UserTokenType, tok.TokenType))]
        if tok.IssuedTokenType or tok.IssuerEndpointUrl:
            result += [
                ('  Issued Token type', tok.IssuedTokenType),
                ('  Issuer Endpoint URL', tok.IssuerEndpointUrl)]
        if tok.SecurityPolicyUri:
            result.append(('  Security Policy URI', tok.SecurityPolicyUri))
    result += [
        ('Transport Profile URI', ep.TransportProfileUri),
        ('Security Level', ep.SecurityLevel)]
    return result


def uadiscover():
    parser = argparse.ArgumentParser(description="Performs OPC UA discovery and prints information on servers and endpoints.")
    add_minimum_args(parser)
    parser.add_argument("-n",
                        "--network",
                        action="store_true",
                        help="Also send a FindServersOnNetwork request to server")
    #parser.add_argument("-s",
                        #"--servers",
                        #action="store_false",
                        #help="send a FindServers request to server")
    #parser.add_argument("-e",
                        #"--endpoints",
                        #action="store_false",
                        #help="send a GetEndpoints request to server")
    args = parser.parse_args()
    logging.basicConfig(format="%(levelname)s: %(message)s", level=getattr(logging, args.loglevel))
    
    if args.network:
        client = Client(args.url, timeout=args.timeout)
        print("Performing discovery at {}\n".format(args.url))
        for i, server in enumerate(client.find_all_servers_on_network(), start=1):
            print('Server {}:'.format(i))
            #for (n, v) in application_to_strings(server):
                #print('  {}: {}'.format(n, v))
            print('')

    client = Client(args.url, timeout=args.timeout)
    print("Performing discovery at {}\n".format(args.url))
    for i, server in enumerate(client.find_all_servers(), start=1):
        print('Server {}:'.format(i))
        for (n, v) in application_to_strings(server):
            print('  {}: {}'.format(n, v))
        print('')

    client = Client(args.url, timeout=args.timeout)
    for i, ep in enumerate(client.get_server_endpoints(), start=1):
        print('Endpoint {}:'.format(i))
        for (n, v) in endpoint_to_strings(ep):
            print('  {}: {}'.format(n, v))
        print('')

    sys.exit(0)


def print_history(o):
    if isinstance(o, ua.HistoryData):
        print("{:30} {:10} {}".format('Source timestamp', 'Status', 'Value'))
        for d in o.DataValues:
            print("{:30} {:10} {}".format(str(d.SourceTimestamp), d.StatusCode.name, d.Value))


def str_to_datetime(s):
    if not s:
        return datetime.utcnow()
    # try different datetime formats
    for fmt in ["%Y-%m-%d", "%Y-%m-%d %H:%M", "%Y-%m-%d %H:%M:%S"]:
        try:
            return datetime.strptime(s, fmt)
        except ValueError:
            pass


def uahistoryread():
    parser = argparse.ArgumentParser(description="Read history of a node")
    add_common_args(parser)
    parser.add_argument("--starttime",
                        default="",
                        help="Start time, formatted as YYYY-MM-DD [HH:MM[:SS]]. Default: current time")
    parser.add_argument("--endtime",
                        default="",
                        help="End time, formatted as YYYY-MM-DD [HH:MM[:SS]]. Default: current time")

    args = parser.parse_args()
    if args.nodeid == "i=84" and args.path == "":
        parser.print_usage()
        print("uahistoryread: error: A NodeId or BrowsePath is required")
        sys.exit(1)
    logging.basicConfig(format="%(levelname)s: %(message)s", level=getattr(logging, args.loglevel))

    client = Client(args.url, timeout=args.timeout)
    client.connect()
    try:
        node = client.get_node(args.nodeid)
        if args.path:
            node = node.get_child(args.path.split(","))
        starttime = str_to_datetime(args.starttime)
        endtime = str_to_datetime(args.endtime)
        print("Reading raw history of node {} at {}; start at {}, end at {}\n".format(node, args.url, starttime, endtime))
        print_history(node.read_raw_history(starttime, endtime))
    finally:
        client.disconnect()
    sys.exit(0)


1			import logging
2			import sys
3			import argparse
4			from datetime import datetime
5			import code
6			from enum import Enum
7
8			from opcua import ua
9			from opcua import Client
10			from opcua import Node
11
12
13			def add_minimum_args(parser):
14			parser.add_argument("-u",
15			"--url",
16			help="URL of OPC UA server (for example: opc.tcp://example.org:4840)",
17			default='opc.tcp://localhost:4841',
18			metavar="URL")
19			parser.add_argument("-v",
20			"--verbose",
21			dest="loglevel",
22			choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'],
23			default='WARNING',
24			help="Set log level")
25			parser.add_argument("--timeout",
26			dest="timeout",
27			type=int,
28			default=1,
29			help="Set socket timeout (NOT the diverse UA timeouts)")
30
31
32			def add_common_args(parser):
33			add_minimum_args(parser)
34			parser.add_argument("-n",
35			"--nodeid",
36			help="Fully-qualified node ID (for example: i=85). Default: root node",
37			default='i=84',
38			metavar="NODE")
39			parser.add_argument("-p",
40			"--path",
41			help="Comma separated browse path to the node starting at nodeid (for example: 3:Mybject,3:MyVariable)",
42			default='',
43			metavar="BROWSEPATH")
44			parser.add_argument("-i",
45			"--namespace",
46			help="Default namespace",
47			type=int,
48			default=0,
49			metavar="NAMESPACE")
50
51
52			def get_node(client, args):
53			node = client.get_node(args.nodeid)
54			if args.path:
55			node = node.get_child(args.path.split(","))
56
57
58			def uaread():
59			parser = argparse.ArgumentParser(description="Read attribute of a node, per default reads value of a node")
60			add_common_args(parser)
61			parser.add_argument("-a",
62			"--attribute",
63			dest="attribute",
64			type=int,
65			default=ua.AttributeIds.Value,
66			help="Set attribute to read")
67			parser.add_argument("-t",
68			"--datatype",
69			dest="datatype",
70			default="python",
71			choices=['python', 'variant', 'datavalue'],
72			help="Data type to return")
73
74			args = parser.parse_args()
75			if args.nodeid == "i=84" and args.path == "" and args.attribute == ua.AttributeIds.Value:
76			parser.print_usage()
77			print("uaread: error: A NodeId or BrowsePath is required")
78			sys.exit(1)
79			logging.basicConfig(format="%(levelname)s: %(message)s", level=getattr(logging, args.loglevel))
80
81			client = Client(args.url, timeout=args.timeout)
82			client.connect()
83			try:
84			node = client.get_node(args.nodeid)
85			if args.path:
86			node = node.get_child(args.path.split(","))
87			attr = node.get_attribute(args.attribute)
88			if args.datatype == "python":
89			print(attr.Value.Value)
90			elif args.datatype == "variant":
91			print(attr.Value)
92			else:
93			print(attr)
94			finally:
95			client.disconnect()
96			sys.exit(0)
97			print(args)
98
99
100			def _args_to_array(val, array):
101			if array == "guess":
102			if "," in val:
103			array = "true"
104			if array == "true":
105			val = val.split(",")
106			return val
107
108
109			def _arg_to_bool(val):
110			if val in ("true", "True"):
111			return True
112			else:
113			return False
114
115
116			def _arg_to_variant(val, array, ptype, varianttype=None):
117			val = _args_to_array(val, array)
118			if isinstance(val, list):
119			val = [ptype(i) for i in val]
120			else:
121			val = ptype(val)
122			if varianttype:
123			return ua.Variant(val, varianttype)
124			else:
125			return ua.Variant(val)
126
127
128			def _val_to_variant(val, args):
129			array = args.array
130			if args.datatype == "guess":
131			if val in ("true", "True", "false", "False"):
132			return _arg_to_variant(val, array, _arg_to_bool)
133			# FIXME: guess bool value
134			try:
135			return _arg_to_variant(val, array, int)
136			except ValueError:
137			try:
138			return _arg_to_variant(val, array, float)
139			except ValueError:
140			return _arg_to_variant(val, array, str)
141			elif args.datatype == "bool":
142			if val in ("1", "True", "true"):
143			return ua.Variant(True, ua.VariantType.Boolean)
144			else:
145			return ua.Variant(False, ua.VariantType.Boolean)
146			elif args.datatype == "sbyte":
147			return _arg_to_variant(val, array, int, ua.VariantType.SByte)
148			elif args.datatype == "byte":
149			return _arg_to_variant(val, array, int, ua.VariantType.Byte)
150			#elif args.datatype == "uint8":
151			#return _arg_to_variant(val, array, int, ua.VariantType.Byte)
152			elif args.datatype == "uint16":
153			return _arg_to_variant(val, array, int, ua.VariantType.UInt16)
154			elif args.datatype == "uint32":
155			return _arg_to_variant(val, array, int, ua.VariantType.UInt32)
156			elif args.datatype == "uint64":
157			return _arg_to_variant(val, array, int, ua.VariantType.UInt64)
158			#elif args.datatype == "int8":
159			#return ua.Variant(int(val), ua.VariantType.Int8)
160			elif args.datatype == "int16":
161			return _arg_to_variant(val, array, int, ua.VariantType.Int16)
162			elif args.datatype == "int32":
163			return _arg_to_variant(val, array, int, ua.VariantType.Int32)
164			elif args.datatype == "int64":
165			return _arg_to_variant(val, array, int, ua.VariantType.Int64)
166			elif args.datatype == "float":
167			return _arg_to_variant(val, array, float, ua.VariantType.Float)
168			elif args.datatype == "double":
169			return _arg_to_variant(val, array, float, ua.VariantType.Double)
170			elif args.datatype == "string":
171			return _arg_to_variant(val, array, str, ua.VariantType.String)
172			elif args.datatype == "datetime":
173			raise NotImplementedError
174			elif args.datatype == "Guid":
175			return _arg_to_variant(val, array, bytes, ua.VariantType.Guid)
176			elif args.datatype == "ByteString":
177			return _arg_to_variant(val, array, bytes, ua.VariantType.ByteString)
178			elif args.datatype == "xml":
179			return _arg_to_variant(val, array, str, ua.VariantType.XmlElement)
180			elif args.datatype == "nodeid":
181			return _arg_to_variant(val, array, ua.NodeId.from_string, ua.VariantType.NodeId)
182			elif args.datatype == "expandednodeid":
183			return _arg_to_variant(val, array, ua.ExpandedNodeId.from_string, ua.VariantType.ExpandedNodeId)
184			elif args.datatype == "statuscode":
185			return _arg_to_variant(val, array, int, ua.VariantType.StatusCode)
186			elif args.datatype in ("qualifiedname", "browsename"):
187			return _arg_to_variant(val, array, ua.QualifiedName.from_string, ua.VariantType.QualifiedName)
188			elif args.datatype == "LocalizedText":
189			return _arg_to_variant(val, array, ua.LocalizedText, ua.VariantTypeLocalizedText)
190
191
192			def uawrite():
193			parser = argparse.ArgumentParser(description="Write attribute of a node, per default write value of node")
194			add_common_args(parser)
195			parser.add_argument("-a",
196			"--attribute",
197			dest="attribute",
198			type=int,
199			default=ua.AttributeIds.Value,
200			help="Set attribute to read")
201			parser.add_argument("-l",
202			"--list",
203			"--array",
204			dest="array",
205			default="guess",
206			choices=["guess", "true", "false"],
207			help="Value is an array")
208			parser.add_argument("-t",
209			"--datatype",
210			dest="datatype",
211			default="guess",
212			choices=["guess", 'byte', 'sbyte', 'nodeid', 'expandednodeid', 'qualifiedname', 'browsename', 'string', 'float', 'double', 'int16', 'int32', "int64", 'uint16', 'uint32', 'uint64', "bool", "string", 'datetime', 'bytestring', 'xmlelement', 'statuscode', 'localizedtext'],
213			help="Data type to return")
214			parser.add_argument("value",
215			help="Value to be written",
216			metavar="VALUE")
217			args = parser.parse_args()
218			if args.nodeid == "i=84" and args.path == "" and args.attribute == ua.AttributeIds.Value:
219			parser.print_usage()
220			print("uaread: error: A NodeId or BrowsePath is required")
221			sys.exit(1)
222			logging.basicConfig(format="%(levelname)s: %(message)s", level=getattr(logging, args.loglevel))
223
224			client = Client(args.url, timeout=args.timeout)
225			client.connect()
226			try:
227			node = client.get_node(args.nodeid)
228			if args.path:
229			node = node.get_child(args.path.split(","))
230			val = _val_to_variant(args.value, args)
231			node.set_attribute(args.attribute, ua.DataValue(val))
232			finally:
233			client.disconnect()
234			sys.exit(0)
235			print(args)
236
237
238			def uals():
239			parser = argparse.ArgumentParser(description="Browse OPC-UA node and print result")
240			add_common_args(parser)
241			parser.add_argument("-l",
242			dest="long_format",
243			const=3,
244			nargs="?",
245			type=int,
246			help="use a long listing format")
247			parser.add_argument("-d",
248			"--depth",
249			default=1,
250			type=int,
251			help="Browse depth")
252
253			args = parser.parse_args()
254			if args.long_format is None:
255			args.long_format = 1
256			logging.basicConfig(format="%(levelname)s: %(message)s", level=getattr(logging, args.loglevel))
257
258			client = Client(args.url, timeout=args.timeout)
259			client.connect()
260			try:
261			node = client.get_node(args.nodeid)
262			if args.path:
263			node = node.get_child(args.path.split(","))
264			print("Browsing node {} at {}\n".format(node, args.url))
265			if args.long_format == 0:
266			_lsprint_0(node, args.depth - 1)
267			elif args.long_format == 1:
268			_lsprint_1(node, args.depth - 1)
269			else:
270			_lsprint_long(node, args.depth - 1)
271			finally:
272			client.disconnect()
273			sys.exit(0)
274			print(args)
275
276
277			def _lsprint_0(node, depth, indent=""):
278			if not indent:
279			print("{:30} {:25}".format("DisplayName", "NodeId"))
280			print("")
281			for desc in node.get_children_descriptions():
282			print("{}{:30} {:25}".format(indent, desc.DisplayName.to_string(), desc.NodeId.to_string()))
283			if depth:
284			_lsprint_0(Node(node.server, desc.NodeId), depth - 1, indent + " ")
285
286
287			def _lsprint_1(node, depth, indent=""):
288			if not indent:
289			print("{:30} {:25} {:25} {:25}".format("DisplayName", "NodeId", "BrowseName", "Value"))
290			print("")
291
292			for desc in node.get_children_descriptions():
293			if desc.NodeClass == ua.NodeClass.Variable:
294			val = Node(node.server, desc.NodeId).get_value()
295			print("{}{:30} {!s:25} {!s:25}, {!s:3}".format(indent, desc.DisplayName.to_string(), desc.NodeId.to_string(), desc.BrowseName.to_string(), val))
296			else:
297			print("{}{:30} {!s:25} {!s:25}".format(indent, desc.DisplayName.to_string(), desc.NodeId.to_string(), desc.BrowseName.to_string()))
298			if depth:
299			_lsprint_1(Node(node.server, desc.NodeId), depth - 1, indent + " ")
300
301
302			def _lsprint_long(pnode, depth, indent=""):
303			if not indent:
304			print("{:30} {:25} {:25} {:10} {:30} {:25}".format("DisplayName", "NodeId", "BrowseName", "DataType", "Timestamp", "Value"))
305			print("")
306			for node in pnode.get_children():
307			attrs = node.get_attributes([ua.AttributeIds.DisplayName,
308			ua.AttributeIds.BrowseName,
309			ua.AttributeIds.NodeClass,
310			ua.AttributeIds.WriteMask,
311			ua.AttributeIds.UserWriteMask,
312			ua.AttributeIds.DataType,
313			ua.AttributeIds.Value])
314			name, bname, nclass, mask, umask, dtype, val = [attr.Value.Value for attr in attrs]
315			update = attrs[-1].ServerTimestamp
316			if nclass == ua.NodeClass.Variable:
317			print("{}{:30} {:25} {:25} {:10} {!s:30} {!s:25}".format(indent, name.to_string(), node.nodeid.to_string(), bname.to_string(), dtype.to_string(), update, val))
318			else:
319			print("{}{:30} {:25} {:25}".format(indent, name.to_string(), bname.to_string(), node.nodeid.to_string()))
320			if depth:
321			_lsprint_long(node, depth - 1, indent + " ")
322
323
324			class SubHandler(object):
325
326			def data_change(self, handle, node, val, attr):
327			print("New data change event", handle, node, val, attr)
328
329			def event(self, handle, event):
330			print("New event", handle, event)
331
332
333			def uasubscribe():
334			parser = argparse.ArgumentParser(description="Subscribe to a node and print results")
335			add_common_args(parser)
336			parser.add_argument("-t",
337			"--eventtype",
338			dest="eventtype",
339			default="datachange",
340			choices=['datachange', 'event'],
341			help="Event type to subscribe to")
342
343			args = parser.parse_args()
344			if args.nodeid == "i=84" and args.path == "":
345			parser.print_usage()
346			print("uaread: error: The NodeId or BrowsePath of a variable is required")
347			sys.exit(1)
348			logging.basicConfig(format="%(levelname)s: %(message)s", level=getattr(logging, args.loglevel))
349
350			client = Client(args.url, timeout=args.timeout)
351			client.connect()
352			try:
353			node = client.get_node(args.nodeid)
354			if args.path:
355			node = node.get_child(args.path.split(","))
356			handler = SubHandler()
357			sub = client.create_subscription(500, handler)
358			if args.eventtype == "datachange":
359			sub.subscribe_data_change(node)
360			else:
361			sub.subscribe_events(node)
362			glbs = globals()
363			glbs.update(locals())
364			shell = code.InteractiveConsole(vars)
365			shell.interact()
366			finally:
367			client.disconnect()
368			sys.exit(0)
369			print(args)
370
371
372			# converts numeric value to its enum name.
373			def enum_to_string(klass, value):
374			if isinstance(value, Enum):
375			return value.name
376			# if value is not a subtype of Enum, try to find a constant
377			# with this value in this class
378			for k, v in vars(klass).items():
379			if not k.startswith('__') and v == value:
380			return k
381			return 'Unknown {} ({})'.format(klass.__name__, value)
382
383
384			def application_to_strings(app):
385			result = []
386			result.append(('Application URI', app.ApplicationUri))
387			optionals = [
388			('Product URI', app.ProductUri),
389			('Application Name', app.ApplicationName.to_string()),
390			('Application Type', enum_to_string(ua.ApplicationType, app.ApplicationType)),
391			('Gateway Server URI', app.GatewayServerUri),
392			('Discovery Profile URI', app.DiscoveryProfileUri),
393			]
394			for (n, v) in optionals:
395			if v:
396			result.append((n, v))
397			for url in app.DiscoveryUrls:
398			result.append(('Discovery URL', url))
399			return result # ['{}: {}'.format(n, v) for (n, v) in result]
400
401
402			def endpoint_to_strings(ep):
403			result = [('Endpoint URL', ep.EndpointUrl)]
404			result += application_to_strings(ep.Server)
405			result += [
406			('Server Certificate', len(ep.ServerCertificate)),
407			('Security Mode', enum_to_string(ua.MessageSecurityMode, ep.SecurityMode)),
408			('Security Policy URI', ep.SecurityPolicyUri)]
409			for tok in ep.UserIdentityTokens:
410			result += [
411			('User policy', tok.PolicyId),
412			(' Token type', enum_to_string(ua.UserTokenType, tok.TokenType))]
413			if tok.IssuedTokenType or tok.IssuerEndpointUrl:
414			result += [
415			(' Issued Token type', tok.IssuedTokenType),
416			(' Issuer Endpoint URL', tok.IssuerEndpointUrl)]
417			if tok.SecurityPolicyUri:
418			result.append((' Security Policy URI', tok.SecurityPolicyUri))
419			result += [
420			('Transport Profile URI', ep.TransportProfileUri),
421			('Security Level', ep.SecurityLevel)]
422			return result
423
424
425			def uadiscover():
426			parser = argparse.ArgumentParser(description="Performs OPC UA discovery and prints information on servers and endpoints.")
427			add_minimum_args(parser)
428			parser.add_argument("-n",
429			"--network",
430			action="store_true",
431			help="Also send a FindServersOnNetwork request to server")
432			#parser.add_argument("-s",
433			#"--servers",
434			#action="store_false",
435			#help="send a FindServers request to server")
436			#parser.add_argument("-e",
437			#"--endpoints",
438			#action="store_false",
439			#help="send a GetEndpoints request to server")
440			args = parser.parse_args()
441			logging.basicConfig(format="%(levelname)s: %(message)s", level=getattr(logging, args.loglevel))
442
443			if args.network:
444			client = Client(args.url, timeout=args.timeout)
445			print("Performing discovery at {}\n".format(args.url))
446			for i, server in enumerate(client.find_all_servers_on_network(), start=1):
447			print('Server {}:'.format(i))
448			#for (n, v) in application_to_strings(server):
449			#print(' {}: {}'.format(n, v))
450			print('')
451
452			client = Client(args.url, timeout=args.timeout)
453			print("Performing discovery at {}\n".format(args.url))
454			for i, server in enumerate(client.find_all_servers(), start=1):
455			print('Server {}:'.format(i))
456			for (n, v) in application_to_strings(server):
457			print(' {}: {}'.format(n, v))
458			print('')
459
460			client = Client(args.url, timeout=args.timeout)
461			for i, ep in enumerate(client.get_server_endpoints(), start=1):
462			print('Endpoint {}:'.format(i))
463			for (n, v) in endpoint_to_strings(ep):
464			print(' {}: {}'.format(n, v))
465			print('')
466
467			sys.exit(0)
468
469
470			def print_history(o):
471			if isinstance(o, ua.HistoryData):
472			print("{:30} {:10} {}".format('Source timestamp', 'Status', 'Value'))
473			for d in o.DataValues:
474			print("{:30} {:10} {}".format(str(d.SourceTimestamp), d.StatusCode.name, d.Value))
475
476
477			def str_to_datetime(s):
478			if not s:
479			return datetime.utcnow()
480			# try different datetime formats
481			for fmt in ["%Y-%m-%d", "%Y-%m-%d %H:%M", "%Y-%m-%d %H:%M:%S"]:
482			try:
483			return datetime.strptime(s, fmt)
484			except ValueError:
485			pass
486
487
488			def uahistoryread():
489			parser = argparse.ArgumentParser(description="Read history of a node")
490			add_common_args(parser)
491			parser.add_argument("--starttime",
492			default="",
493			help="Start time, formatted as YYYY-MM-DD [HH:MM[:SS]]. Default: current time")
494			parser.add_argument("--endtime",
495			default="",
496			help="End time, formatted as YYYY-MM-DD [HH:MM[:SS]]. Default: current time")
497
498			args = parser.parse_args()
499			if args.nodeid == "i=84" and args.path == "":
500			parser.print_usage()
501			print("uahistoryread: error: A NodeId or BrowsePath is required")
502			sys.exit(1)
503			logging.basicConfig(format="%(levelname)s: %(message)s", level=getattr(logging, args.loglevel))
504
505			client = Client(args.url, timeout=args.timeout)
506			client.connect()
507			try:
508			node = client.get_node(args.nodeid)
509			if args.path:
510			node = node.get_child(args.path.split(","))
511			starttime = str_to_datetime(args.starttime)
512			endtime = str_to_datetime(args.endtime)
513			print("Reading raw history of node {} at {}; start at {}, end at {}\n".format(node, args.url, starttime, endtime))
514			print_history(node.read_raw_history(starttime, endtime))
515			finally:
516			client.disconnect()
517			sys.exit(0)
518

FreeOpcUa / python-opcua

Push — dev ( a35960...17cd27 )

opcua.uadiscover() C

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like