opcua.uaread() - Code Metrics - Inspection of "small README update" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — dev ( 3b343f...d958a8 )

by Olivier

created 2015-12-10 21:08 UTC

opcua.uaread() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	3
dl	0
loc	33
rs	8.8571

import logging
import sys
import argparse
from datetime import datetime
from enum import Enum
import math
import time

try:
    from IPython import embed
except ImportError:
    import code

    def embed():
        code.interact(local=dict(globals(), **locals())) 

from opcua import ua
from opcua import Client
from opcua import Server
from opcua import Node
from opcua import uamethod


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 NODE (for example: 3:Mybject,3:MyVariable)",
                        default='',
                        metavar="BROWSEPATH")
    parser.add_argument("-i",
                        "--namespace",
                        help="Default namespace",
                        type=int,
                        default=0,
                        metavar="NAMESPACE")


def parse_args(parser, requirenodeid=False):
    args = parser.parse_args()
    logging.basicConfig(format="%(levelname)s: %(message)s", level=getattr(logging, args.loglevel))
    if args.url and '://' not in args.url:
        logging.info("Adding default scheme %s to URL %s", ua.OPC_TCP_SCHEME, args.url)
        args.url = ua.OPC_TCP_SCHEME + '://' + args.url
    # check that a nodeid has been given explicitly, a bit hackish...
    if requirenodeid and args.nodeid == "i=84" and args.path == "":
        parser.print_usage()
        print("{}: error: A NodeId or BrowsePath is required".format(parser.prog))
        sys.exit(1)
    return args


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


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 = parse_args(parser, requirenodeid=True)

    client = Client(args.url, timeout=args.timeout)
    client.connect()
    try:
        node = get_node(client, args)
        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 = parse_args(parser, requirenodeid=True)

    client = Client(args.url, timeout=args.timeout)
    client.connect()
    try:
        node = get_node(client, args)
        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 = parse_args(parser)
    if args.long_format is None:
        args.long_format = 1

    client = Client(args.url, timeout=args.timeout)
    client.connect()
    try:
        node = get_node(client, args)
        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 = parse_args(parser, requirenodeid=True)

    client = Client(args.url, timeout=args.timeout)
    client.connect()
    try:
        node = get_node(client, args)
        handler = SubHandler()
        sub = client.create_subscription(500, handler)
        if args.eventtype == "datachange":
            sub.subscribe_data_change(node)
        else:
            sub.subscribe_events(node)
        embed()
    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 uaclient():
    parser = argparse.ArgumentParser(description="Connect to server and start python shell. root and objects nodes are available. Node specificed in command line is available as mynode variable")
    add_common_args(parser)
    parser.add_argument("-c",
                        "--certificate",
                        help="set client certificate")
    parser.add_argument("-k",
                        "--private_key",
                        help="set client private key")
    args = parse_args(parser)

    client = Client(args.url, timeout=args.timeout)
    client.connect()
    if args.certificate:
        client.load_certificate(args.certificate)
    if args.private_key:
        client.load_certificate(args.private_key)
    try:
        root = client.get_root_node()
        objects = client.get_objects_node()
        mynode = get_node(client, args)
        embed()
    finally:
        client.disconnect()
    sys.exit(0)


def uaserver():
    parser = argparse.ArgumentParser(description="Run an example OPC-UA server. By importing xml definition and using uawrite command line, it is even possible to expose real data using this server")
    # we setup a server, this is a bit different from other tool so we do not reuse common arguments
    parser.add_argument("-u",
                        "--url",
                        help="URL of OPC UA server, default is opc.tcp://0.0.0.0:4841",
                        default='opc.tcp://0.0.0.0: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("-x",
                        "--xml",
                        metavar="XML_FILE",
                        help="Populate address space with nodes defined in XML")
    parser.add_argument("-p",
                        "--populate",
                        action="store_false",
                        help="Populate address space with some sample nodes")
    parser.add_argument("-s",
                        "--shell",
                        action="store_true",
                        help="Start python shell instead of randomly changing node values")
    args = parser.parse_args()
    logging.basicConfig(format="%(levelname)s: %(message)s", level=getattr(logging, args.loglevel))

    server = Server()
    server.set_endpoint(args.url)
    server.set_server_name("FreeOpcUa Example Server")
    if args.xml:
        server.import_xml(args.xml)
    if args.populate:
        @uamethod
        def multiply(parent, x, y):
            print("multiply method call with parameters: ", x, y)
            return x * y

        uri = "http://examples.freeopcua.github.io"
        idx = server.register_namespace(uri)
        objects = server.get_objects_node()
        myobj = objects.add_object(idx, "MyObject")
        mywritablevar = myobj.add_variable(idx, "MyWritableVariable", 6.7)
        mywritablevar.set_writable()    # Set MyVariable to be writable by clients
        myvar = myobj.add_variable(idx, "MyVariable", 6.7)
        myarrayvar = myobj.add_variable(idx, "MyVarArray", [6.7, 7.9])
        myprop = myobj.add_property(idx, "MyProperty", "I am a property")
        mymethod = myobj.add_method(idx, "MyMethod", multiply, [ua.VariantType.Double, ua.VariantType.Int64], [ua.VariantType.Double])

    server.start()
    try:
        if args.shell:
            embed()
        else:
            count = 0
            while True:
                time.sleep(1)
                myvar.set_value(math.sin(count / 10))
                myarrayvar.set_value([math.sin(count / 10), math.sin(count / 100)])
                count += 1
    finally:
        server.stop()
    sys.exit(0)





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 = parse_args(parser)
    
    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 = parse_args(parser, requirenodeid=True)

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

FreeOpcUa / python-opcua

Push — dev ( 3b343f...d958a8 )

opcua.uaread() B

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like