opcua.str_to_datetime() - Code Metrics - Inspection of "Merge pull request #58 from alkor/tools" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 18f8bd...e443b4 )

by Olivier

created 2015-11-30 11:33 UTC

opcua.str_to_datetime() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
dl	0
loc	9
rs	9.2
cc	4

import logging
import sys
import argparse
from opcua import ua, Client
import code
from enum import Enum
from datetime import datetime


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",
                        #default=ua.AttributeIds.Value,
                        #help="use a long listing format")
    parser.add_argument("-d",
                        "--depth",
                        default=1,
                        type=int,
                        help="Browse depth")

    args = parser.parse_args()
    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))
        _lsprint(client, node.nodeid, args.depth - 1)

    finally:
        client.disconnect()
    sys.exit(0)
    print(args)


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

FreeOpcUa / python-opcua

Push — master ( 18f8bd...e443b4 )

opcua.str_to_datetime() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like