opcua.application_to_strings() - Code Metrics - Inspection of "make use of new enum in tools" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — dev ( 454bf6...555c2d )

by Olivier

created 2016-01-01 15:39 UTC

opcua.application_to_strings() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
dl	0
loc	16
rs	9.2
cc	4

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
from opcua import security_policies


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, default_node='i=84'):
    add_minimum_args(parser)
    parser.add_argument("-n",
                        "--nodeid",
                        help="Fully-qualified node ID (for example: i=85). Default: root node",
                        default=default_node,
                        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")
    parser.add_argument("--security",
                        help="Security settings, for example: Basic256,SignAndEncrypt,cert.der,pk.pem[,server_cert.der]. Default: None",
                        default='')


def _require_nodeid(parser, args):
    # check that a nodeid has been given explicitly, a bit hackish...
    if args.nodeid == "i=84" and args.path == "":
        parser.print_usage()
        print("{}: error: A NodeId or BrowsePath is required".format(parser.prog))
        sys.exit(1)


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
    if requirenodeid:
        _require_nodeid(parser, args)
    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.set_security_string(args.security)
    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)
        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.VariantType.LocalizedText)


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.set_security_string(args.security)
    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.set_security_string(args.security)
    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 datachange_notification(self, node, val, data):
        print("New data change event", node, val, data)

    def event_notification(self, event):
        print("New event", 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=False)
    if args.eventtype == "datachange":
        _require_nodeid(parser, args)
    else:
        # FIXME: this is broken, someone may have written i=84 on purpose
        if args.nodeid == "i=84" and args.path == "":
            args.nodeid = "i=2253"

    client = Client(args.url, timeout=args.timeout)
    client.set_security_string(args.security)
    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)


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', str(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', str(ep.SecurityMode)),
        ('Security Policy URI', ep.SecurityPolicyUri)]
    for tok in ep.UserIdentityTokens:
        result += [
            ('User policy', tok.PolicyId),
            ('  Token type', str(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.set_security_string(args.security)
    client.connect()
    if args.certificate:
        client.load_client_certificate(args.certificate)
    if args.private_key:
        client.load_private_key(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("-c",
                        "--disable-clock",
                        action="store_true",
                        help="Disable clock, to avoid seeing many write if debugging an application")
    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.disable_clock(args.disable_clock)
    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)
    
    client = Client(args.url, timeout=args.timeout)

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

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

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

FreeOpcUa / python-opcua

Push — dev ( 454bf6...555c2d )

opcua.application_to_strings() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like