opcua._lsprint_0() - Code Metrics - Inspection of "fix automatic build and crypto dependencies" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Branch — dev (949ba8)

by Olivier

created 2015-12-02 08:22 UTC

opcua._lsprint_0() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	4
dl	0
loc	8
rs	9.2

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="VALUE",
                        #choices=['VALUE', 'NODEID', 'BROWSENAME', 'ERROR', 'CRITICAL'],
                        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="VALUE",
                        #choices=['VALUE', 'NODEID', 'BROWSENAME', 'ERROR', 'CRITICAL'],
                        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",
                        #action="store_true",
                        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)
    args = parser.parse_args()
    logging.basicConfig(format="%(levelname)s: %(message)s", level=getattr(logging, args.loglevel))

    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="VALUE",
66			#choices=['VALUE', 'NODEID', 'BROWSENAME', 'ERROR', 'CRITICAL'],
67			default=ua.AttributeIds.Value,
68			help="Set attribute to read")
69			parser.add_argument("-t",
70			"--datatype",
71			dest="datatype",
72			default="python",
73			choices=['python', 'variant', 'datavalue'],
74			help="Data type to return")
75
76			args = parser.parse_args()
77			if args.nodeid == "i=84" and args.path == "" and args.attribute == ua.AttributeIds.Value:
78			parser.print_usage()
79			print("uaread: error: A NodeId or BrowsePath is required")
80			sys.exit(1)
81			logging.basicConfig(format="%(levelname)s: %(message)s", level=getattr(logging, args.loglevel))
82
83			client = Client(args.url, timeout=args.timeout)
84			client.connect()
85			try:
86			node = client.get_node(args.nodeid)
87			if args.path:
88			node = node.get_child(args.path.split(","))
89			attr = node.get_attribute(args.attribute)
90			if args.datatype == "python":
91			print(attr.Value.Value)
92			elif args.datatype == "variant":
93			print(attr.Value)
94			else:
95			print(attr)
96			finally:
97			client.disconnect()
98			sys.exit(0)
99			print(args)
100
101
102			def _args_to_array(val, array):
103			if array == "guess":
104			if "," in val:
105			array = "true"
106			if array == "true":
107			val = val.split(",")
108			return val
109
110
111			def _arg_to_bool(val):
112			if val in ("true", "True"):
113			return True
114			else:
115			return False
116
117
118			def _arg_to_variant(val, array, ptype, varianttype=None):
119			val = _args_to_array(val, array)
120			if isinstance(val, list):
121			val = [ptype(i) for i in val]
122			else:
123			val = ptype(val)
124			if varianttype:
125			return ua.Variant(val, varianttype)
126			else:
127			return ua.Variant(val)
128
129
130			def _val_to_variant(val, args):
131			array = args.array
132			if args.datatype == "guess":
133			if val in ("true", "True", "false", "False"):
134			return _arg_to_variant(val, array, _arg_to_bool)
135			# FIXME: guess bool value
136			try:
137			return _arg_to_variant(val, array, int)
138			except ValueError:
139			try:
140			return _arg_to_variant(val, array, float)
141			except ValueError:
142			return _arg_to_variant(val, array, str)
143			elif args.datatype == "bool":
144			if val in ("1", "True", "true"):
145			return ua.Variant(True, ua.VariantType.Boolean)
146			else:
147			return ua.Variant(False, ua.VariantType.Boolean)
148			elif args.datatype == "sbyte":
149			return _arg_to_variant(val, array, int, ua.VariantType.SByte)
150			elif args.datatype == "byte":
151			return _arg_to_variant(val, array, int, ua.VariantType.Byte)
152			#elif args.datatype == "uint8":
153			#return _arg_to_variant(val, array, int, ua.VariantType.Byte)
154			elif args.datatype == "uint16":
155			return _arg_to_variant(val, array, int, ua.VariantType.UInt16)
156			elif args.datatype == "uint32":
157			return _arg_to_variant(val, array, int, ua.VariantType.UInt32)
158			elif args.datatype == "uint64":
159			return _arg_to_variant(val, array, int, ua.VariantType.UInt64)
160			#elif args.datatype == "int8":
161			#return ua.Variant(int(val), ua.VariantType.Int8)
162			elif args.datatype == "int16":
163			return _arg_to_variant(val, array, int, ua.VariantType.Int16)
164			elif args.datatype == "int32":
165			return _arg_to_variant(val, array, int, ua.VariantType.Int32)
166			elif args.datatype == "int64":
167			return _arg_to_variant(val, array, int, ua.VariantType.Int64)
168			elif args.datatype == "float":
169			return _arg_to_variant(val, array, float, ua.VariantType.Float)
170			elif args.datatype == "double":
171			return _arg_to_variant(val, array, float, ua.VariantType.Double)
172			elif args.datatype == "string":
173			return _arg_to_variant(val, array, str, ua.VariantType.String)
174			elif args.datatype == "datetime":
175			raise NotImplementedError
176			elif args.datatype == "Guid":
177			return _arg_to_variant(val, array, bytes, ua.VariantType.Guid)
178			elif args.datatype == "ByteString":
179			return _arg_to_variant(val, array, bytes, ua.VariantType.ByteString)
180			elif args.datatype == "xml":
181			return _arg_to_variant(val, array, str, ua.VariantType.XmlElement)
182			elif args.datatype == "nodeid":
183			return _arg_to_variant(val, array, ua.NodeId.from_string, ua.VariantType.NodeId)
184			elif args.datatype == "expandednodeid":
185			return _arg_to_variant(val, array, ua.ExpandedNodeId.from_string, ua.VariantType.ExpandedNodeId)
186			elif args.datatype == "statuscode":
187			return _arg_to_variant(val, array, int, ua.VariantType.StatusCode)
188			elif args.datatype in ("qualifiedname", "browsename"):
189			return _arg_to_variant(val, array, ua.QualifiedName.from_string, ua.VariantType.QualifiedName)
190			elif args.datatype == "LocalizedText":
191			return _arg_to_variant(val, array, ua.LocalizedText, ua.VariantTypeLocalizedText)
192
193
194			def uawrite():
195			parser = argparse.ArgumentParser(description="Write attribute of a node, per default write value of node")
196			add_common_args(parser)
197			parser.add_argument("-a",
198			"--attribute",
199			dest="attribute",
200			type=int,
201			#default="VALUE",
202			#choices=['VALUE', 'NODEID', 'BROWSENAME', 'ERROR', 'CRITICAL'],
203			default=ua.AttributeIds.Value,
204			help="Set attribute to read")
205			parser.add_argument("-l",
206			"--list",
207			"--array",
208			dest="array",
209			default="guess",
210			choices=["guess", "true", "false"],
211			help="Value is an array")
212			parser.add_argument("-t",
213			"--datatype",
214			dest="datatype",
215			default="guess",
216			choices=["guess", 'byte', 'sbyte', 'nodeid', 'expandednodeid', 'qualifiedname', 'browsename', 'string', 'float', 'double', 'int16', 'int32', "int64", 'uint16', 'uint32', 'uint64', "bool", "string", 'datetime', 'bytestring', 'xmlelement', 'statuscode', 'localizedtext'],
217			help="Data type to return")
218			parser.add_argument("value",
219			help="Value to be written",
220			metavar="VALUE")
221			args = parser.parse_args()
222			if args.nodeid == "i=84" and args.path == "" and args.attribute == ua.AttributeIds.Value:
223			parser.print_usage()
224			print("uaread: error: A NodeId or BrowsePath is required")
225			sys.exit(1)
226			logging.basicConfig(format="%(levelname)s: %(message)s", level=getattr(logging, args.loglevel))
227
228			client = Client(args.url, timeout=args.timeout)
229			client.connect()
230			try:
231			node = client.get_node(args.nodeid)
232			if args.path:
233			node = node.get_child(args.path.split(","))
234			val = _val_to_variant(args.value, args)
235			node.set_attribute(args.attribute, ua.DataValue(val))
236			finally:
237			client.disconnect()
238			sys.exit(0)
239			print(args)
240
241
242			def uals():
243			parser = argparse.ArgumentParser(description="Browse OPC-UA node and print result")
244			add_common_args(parser)
245			parser.add_argument("-l",
246			dest="long_format",
247			#action="store_true",
248			const=3,
249			nargs="?",
250			type=int,
251			help="use a long listing format")
252			parser.add_argument("-d",
253			"--depth",
254			default=1,
255			type=int,
256			help="Browse depth")
257
258			args = parser.parse_args()
259			if args.long_format is None:
260			args.long_format = 1
261			logging.basicConfig(format="%(levelname)s: %(message)s", level=getattr(logging, args.loglevel))
262
263			client = Client(args.url, timeout=args.timeout)
264			client.connect()
265			try:
266			node = client.get_node(args.nodeid)
267			if args.path:
268			node = node.get_child(args.path.split(","))
269			print("Browsing node {} at {}\n".format(node, args.url))
270			if args.long_format == 0:
271			_lsprint_0(node, args.depth - 1)
272			elif args.long_format == 1:
273			_lsprint_1(node, args.depth - 1)
274			else:
275			_lsprint_long(node, args.depth - 1)
276			finally:
277			client.disconnect()
278			sys.exit(0)
279			print(args)
280
281
282			def _lsprint_0(node, depth, indent=""):
283			if not indent:
284			print("{:30} {:25}".format("DisplayName", "NodeId"))
285			print("")
286			for desc in node.get_children_descriptions():
287			print("{}{:30} {:25}".format(indent, desc.DisplayName.to_string(), desc.NodeId.to_string()))
288			if depth:
289			_lsprint_0(Node(node.server, desc.NodeId), depth - 1, indent + " ")
290
291
292			def _lsprint_1(node, depth, indent=""):
293			if not indent:
294			print("{:30} {:25} {:25} {:25}".format("DisplayName", "NodeId", "BrowseName", "Value"))
295			print("")
296
297			for desc in node.get_children_descriptions():
298			if desc.NodeClass == ua.NodeClass.Variable:
299			val = Node(node.server, desc.NodeId).get_value()
300			print("{}{:30} {!s:25} {!s:25}, {!s:3}".format(indent, desc.DisplayName.to_string(), desc.NodeId.to_string(), desc.BrowseName.to_string(), val))
301			else:
302			print("{}{:30} {!s:25} {!s:25}".format(indent, desc.DisplayName.to_string(), desc.NodeId.to_string(), desc.BrowseName.to_string()))
303			if depth:
304			_lsprint_1(Node(node.server, desc.NodeId), depth - 1, indent + " ")
305
306
307			def _lsprint_long(pnode, depth, indent=""):
308			if not indent:
309			print("{:30} {:25} {:25} {:10} {:30} {:25}".format("DisplayName", "NodeId", "BrowseName", "DataType", "Timestamp", "Value"))
310			print("")
311			for node in pnode.get_children():
312			attrs = node.get_attributes([ua.AttributeIds.DisplayName,
313			ua.AttributeIds.BrowseName,
314			ua.AttributeIds.NodeClass,
315			ua.AttributeIds.WriteMask,
316			ua.AttributeIds.UserWriteMask,
317			ua.AttributeIds.DataType,
318			ua.AttributeIds.Value])
319			name, bname, nclass, mask, umask, dtype, val = [attr.Value.Value for attr in attrs]
320			update = attrs[-1].ServerTimestamp
321			if nclass == ua.NodeClass.Variable:
322			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))
323			else:
324			print("{}{:30} {:25} {:25}".format(indent, name.to_string(), bname.to_string(), node.nodeid.to_string()))
325			if depth:
326			_lsprint_long(node, depth - 1, indent + " ")
327
328
329			class SubHandler(object):
330
331			def data_change(self, handle, node, val, attr):
332			print("New data change event", handle, node, val, attr)
333
334			def event(self, handle, event):
335			print("New event", handle, event)
336
337
338			def uasubscribe():
339			parser = argparse.ArgumentParser(description="Subscribe to a node and print results")
340			add_common_args(parser)
341			parser.add_argument("-t",
342			"--eventtype",
343			dest="eventtype",
344			default="datachange",
345			choices=['datachange', 'event'],
346			help="Event type to subscribe to")
347
348			args = parser.parse_args()
349			if args.nodeid == "i=84" and args.path == "":
350			parser.print_usage()
351			print("uaread: error: The NodeId or BrowsePath of a variable is required")
352			sys.exit(1)
353			logging.basicConfig(format="%(levelname)s: %(message)s", level=getattr(logging, args.loglevel))
354
355			client = Client(args.url, timeout=args.timeout)
356			client.connect()
357			try:
358			node = client.get_node(args.nodeid)
359			if args.path:
360			node = node.get_child(args.path.split(","))
361			handler = SubHandler()
362			sub = client.create_subscription(500, handler)
363			if args.eventtype == "datachange":
364			sub.subscribe_data_change(node)
365			else:
366			sub.subscribe_events(node)
367			glbs = globals()
368			glbs.update(locals())
369			shell = code.InteractiveConsole(vars)
370			shell.interact()
371			finally:
372			client.disconnect()
373			sys.exit(0)
374			print(args)
375
376
377			# converts numeric value to its enum name.
378			def enum_to_string(klass, value):
379			if isinstance(value, Enum):
380			return value.name
381			# if value is not a subtype of Enum, try to find a constant
382			# with this value in this class
383			for k, v in vars(klass).items():
384			if not k.startswith('__') and v == value:
385			return k
386			return 'Unknown {} ({})'.format(klass.__name__, value)
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', enum_to_string(ua.ApplicationType, 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', enum_to_string(ua.MessageSecurityMode, ep.SecurityMode)),
413			('Security Policy URI', ep.SecurityPolicyUri)]
414			for tok in ep.UserIdentityTokens:
415			result += [
416			('User policy', tok.PolicyId),
417			(' Token type', enum_to_string(ua.UserTokenType, 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 uadiscover():
431			parser = argparse.ArgumentParser(description="Performs OPC UA discovery and prints information on servers and endpoints.")
432			add_minimum_args(parser)
433			args = parser.parse_args()
434			logging.basicConfig(format="%(levelname)s: %(message)s", level=getattr(logging, args.loglevel))
435
436			client = Client(args.url, timeout=args.timeout)
437			print("Performing discovery at {}\n".format(args.url))
438			for i, server in enumerate(client.find_all_servers(), start=1):
439			print('Server {}:'.format(i))
440			for (n, v) in application_to_strings(server):
441			print(' {}: {}'.format(n, v))
442			print('')
443
444			client = Client(args.url, timeout=args.timeout)
445			for i, ep in enumerate(client.get_server_endpoints(), start=1):
446			print('Endpoint {}:'.format(i))
447			for (n, v) in endpoint_to_strings(ep):
448			print(' {}: {}'.format(n, v))
449			print('')
450
451			sys.exit(0)
452
453
454			def print_history(o):
455			if isinstance(o, ua.HistoryData):
456			print("{:30} {:10} {}".format('Source timestamp', 'Status', 'Value'))
457			for d in o.DataValues:
458			print("{:30} {:10} {}".format(str(d.SourceTimestamp), d.StatusCode.name, d.Value))
459
460
461			def str_to_datetime(s):
462			if not s:
463			return datetime.utcnow()
464			# try different datetime formats
465			for fmt in ["%Y-%m-%d", "%Y-%m-%d %H:%M", "%Y-%m-%d %H:%M:%S"]:
466			try:
467			return datetime.strptime(s, fmt)
468			except ValueError:
469			pass
470
471
472			def uahistoryread():
473			parser = argparse.ArgumentParser(description="Read history of a node")
474			add_common_args(parser)
475			parser.add_argument("--starttime",
476			default="",
477			help="Start time, formatted as YYYY-MM-DD [HH:MM[:SS]]. Default: current time")
478			parser.add_argument("--endtime",
479			default="",
480			help="End time, formatted as YYYY-MM-DD [HH:MM[:SS]]. Default: current time")
481
482			args = parser.parse_args()
483			if args.nodeid == "i=84" and args.path == "":
484			parser.print_usage()
485			print("uahistoryread: error: A NodeId or BrowsePath is required")
486			sys.exit(1)
487			logging.basicConfig(format="%(levelname)s: %(message)s", level=getattr(logging, args.loglevel))
488
489			client = Client(args.url, timeout=args.timeout)
490			client.connect()
491			try:
492			node = client.get_node(args.nodeid)
493			if args.path:
494			node = node.get_child(args.path.split(","))
495			starttime = str_to_datetime(args.starttime)
496			endtime = str_to_datetime(args.endtime)
497			print("Reading raw history of node {} at {}; start at {}, end at {}\n".format(node, args.url, starttime, endtime))
498			print_history(node.read_raw_history(starttime, endtime))
499			finally:
500			client.disconnect()
501			sys.exit(0)
502

FreeOpcUa / python-opcua

Branch — dev (949ba8)

opcua._lsprint_0() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like