asyncua.common.events.Event.from_field_dict() - Code Metrics - Inspection of "[WIP] Alarm conditions1" - FreeOpcUa/opcua-asyncio - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#365)

unknown

created 2020-12-18 11:22 UTC

asyncua.common.events.Event.from_field_dict() A

↳ Parent: asyncua.common.events

Complexity

Conditions

Size

Total Lines	9
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	6
nop	1
dl	0
loc	9
rs	10
c	0
b	0
f	0

import copy

from asyncua import ua
import asyncua
from ..ua.uaerrors import UaError
from .ua_utils import get_node_subtypes


class Event:
    """
    OPC UA Event object.
    This is class in inherited by the common event objects such as BaseEvent,
    other auto standard events and custom events
    Events are used to trigger events on server side and are
    sent to clients for every events from server

    Developer Warning:
    On server side the data type of attributes should be known, thus
    add properties using the add_property method!!!
    """

    def __init__(self, emitting_node=ua.ObjectIds.Server):
        self.server_handle = None
        self.select_clauses = None
        self.event_fields = None
        self.data_types = {}
        self.emitting_node = emitting_node
        if isinstance(emitting_node, ua.NodeId):
            self.emitting_node = emitting_node
        else:
            self.emitting_node = ua.NodeId(emitting_node)
        # save current attributes
        self.internal_properties = list(self.__dict__.keys())[:] + ["internal_properties"]

    def __str__(self):
        return "{0}({1})".format(
            self.__class__.__name__,
            [str(k) + ":" + str(v) for k, v in self.__dict__.items() if k not in self.internal_properties])
    __repr__ = __str__

    def add_property(self, name, val, datatype):
        """
        Add a property to event and store its data type
        """
        setattr(self, name, val)
        self.data_types[name] = datatype

    def add_variable(self, name, val, datatype):
        """
        Add a variable to event and store its data type
        variables are able to have properties as children
        """
        setattr(self, name, val)
        self.data_types[name] = datatype

    def get_event_props_as_fields_dict(self):
        """
        convert all properties and variables of the Event class to a dict of variants
        """
        field_vars = {}
        for key, value in vars(self).items():
            if not key.startswith("__") and key not in self.internal_properties:
                field_vars[key] = ua.Variant(value, self.data_types[key])
        return field_vars

    @staticmethod
    def from_field_dict(fields):
        """
        Create an Event object from a dict of name and variants
        """
        ev = Event()
        for k, v in fields.items():
            ev.add_property(k, v.Value, v.VariantType)
        return ev

    def to_event_fields_using_subscription_fields(self, select_clauses):
        """
        Using a new select_clauses and the original select_clauses
        used during subscription, return a field list
        """
        fields = []
        for sattr in select_clauses:
            for idx, o_sattr in enumerate(self.select_clauses):
                if sattr.BrowsePath == o_sattr.BrowsePath and sattr.AttributeId == o_sattr.AttributeId:
                    fields.append(self.event_fields[idx])
                    break
        return fields

    def to_event_fields(self, select_clauses):
        """
        return a field list using a select clause and the object properties
        """
        fields = []
        for sattr in select_clauses:
            if not sattr.BrowsePath:
                name = ua.AttributeIds(sattr.AttributeId).name
            else:
                name = sattr.BrowsePath[0].Name
                iter_paths = iter(sattr.BrowsePath)
                next(iter_paths)
                for path in iter_paths:
                    name += '/' + path.Name
            try:
                val = getattr(self, name)
            except AttributeError:
                field = ua.Variant(None)
            else:
                if val is None:
                    field = ua.Variant(None)
                else:
                    field = ua.Variant(copy.deepcopy(val), self.data_types[name])
            fields.append(field)
        return fields

    @staticmethod
    def from_event_fields(select_clauses, fields):
        """
        Instantiate an Event object from a select_clauses and fields
        """
        ev = Event()
        ev.select_clauses = select_clauses
        ev.event_fields = fields
        for idx, sattr in enumerate(select_clauses):
            if len(sattr.BrowsePath) == 0:
                name = sattr.AttributeId.name
            else:
                name = sattr.BrowsePath[0].Name
                iter_paths = iter(sattr.BrowsePath)
                next(iter_paths)
                for path in iter_paths:
                    name += '/' + path.Name
            ev.add_property(name, fields[idx].Value, fields[idx].VariantType)
        return ev


async def get_filter_from_event_type(eventtypes):
    evfilter = ua.EventFilter()
    evfilter.SelectClauses = await select_clauses_from_evtype(eventtypes)
    evfilter.WhereClause = await where_clause_from_evtype(eventtypes)
    return evfilter


async def select_clauses_from_evtype(evtypes):
    clauses = []
    selected_paths = []
    for evtype in evtypes:
        event_props_and_vars = await get_event_properties_from_type_node(evtype)
        event_props_and_vars.extend(await get_event_variables_from_type_node(evtype))
        for node in event_props_and_vars:
            browse_name = await node.read_browse_name()
            if browse_name not in selected_paths:
                op = ua.SimpleAttributeOperand()
                op.AttributeId = ua.AttributeIds.Value
                op.BrowsePath = [browse_name]
                clauses.append(op)
                selected_paths.append(browse_name)
        for var in await get_event_variables_from_type_node(evtype):
            browse_name = await var.get_browse_name()
            if browse_name not in selected_paths:
                op = ua.SimpleAttributeOperand()
                op.AttributeId = ua.AttributeIds.Value
                op.BrowsePath = [browse_name]
                clauses.append(op)
                selected_paths.append(browse_name)
            for prop in await var.get_properties():
                browse_path = [browse_name, await prop.get_browse_name()]
                if browse_path not in selected_paths:
                    op = ua.SimpleAttributeOperand()
                    op.AttributeId = ua.AttributeIds.Value
                    op.BrowsePath = browse_path
                    clauses.append(op)
                    selected_paths.append(browse_path)
    return clauses


async def where_clause_from_evtype(evtypes):
    cf = ua.ContentFilter()
    el = ua.ContentFilterElement()
    # operands can be ElementOperand, LiteralOperand, AttributeOperand, SimpleAttribute
    # Create a clause where the generate event type property EventType
    # must be a subtype of events in evtypes argument

    # the first operand is the attribute event type
    op = ua.SimpleAttributeOperand()
    # op.TypeDefinitionId = evtype.nodeid
    op.BrowsePath.append(ua.QualifiedName("EventType", 0))
    op.AttributeId = ua.AttributeIds.Value
    el.FilterOperands.append(op)
    # now create a list of all subtypes we want to accept
    subtypes = []
    for evtype in evtypes:
        for st in await get_node_subtypes(evtype):
            subtypes.append(st.nodeid)
    subtypes = list(set(subtypes))  # remove duplicates
    for subtypeid in subtypes:
        op = ua.LiteralOperand()
        op.Value = ua.Variant(subtypeid)
        el.FilterOperands.append(op)
    el.FilterOperator = ua.FilterOperator.InList
    cf.Elements.append(el)
    return cf


async def get_event_variables_from_type_node(node):

    variables = []
    curr_node = node
    while True:
        variables.extend(await curr_node.get_variables())
        if curr_node.nodeid.Identifier == ua.ObjectIds.BaseEventType:
            break
        parents = await curr_node.get_referenced_nodes(
            refs=ua.ObjectIds.HasSubtype, direction=ua.BrowseDirection.Inverse, includesubtypes=True)
        if len(parents) != 1:  # Something went wrong
            return None
        curr_node = parents[0]
    return variables


async def get_event_properties_from_type_node(node):
    properties = []
    curr_node = node
    while True:
        properties.extend(await curr_node.get_properties())
        if curr_node.nodeid.Identifier == ua.ObjectIds.BaseEventType:
            break
        parents = await curr_node.get_referenced_nodes(
            refs=ua.ObjectIds.HasSubtype, direction=ua.BrowseDirection.Inverse, includesubtypes=True
        )
        if len(parents) != 1:  # Something went wrong
            return None
        curr_node = parents[0]
    return properties


async def get_event_variables_from_type_node(node):

    variables = []
    curr_node = node
    while True:
        variables.extend(await curr_node.get_variables())
        if curr_node.nodeid.Identifier == ua.ObjectIds.BaseEventType:
            break
        parents = await curr_node.get_referenced_nodes(
            refs=ua.ObjectIds.HasSubtype, direction=ua.BrowseDirection.Inverse, includesubtypes=True
        )
        if len(parents) != 1:  # Something went wrong
            return None
        curr_node = parents[0]
    return variables


async def get_event_obj_from_type_node(node):
    """
    return an Event object from an event type node
    """
    if node.nodeid.NamespaceIndex == 0 and \
            node.nodeid.Identifier in asyncua.common.event_objects.IMPLEMENTED_EVENTS.keys():
            return asyncua.common.event_objects.IMPLEMENTED_EVENTS[node.nodeid.Identifier]()
    else:
        parent_identifier, parent_eventtype = await _find_parent_eventtype(node)

        class CustomEvent(parent_eventtype):

            def __init__(self):
                parent_eventtype.__init__(self)

                self.EventType = node.nodeid

            async def init(self):
                curr_node = node
                while curr_node.nodeid.Identifier != parent_identifier:

                    for prop in await curr_node.get_properties():
                        name = (await prop.read_browse_name()).Name
                        val = await prop.read_data_value()
                        self.add_property(name, val.Value.Value, val.Value.VariantType)
                    for var in await curr_node.get_variables():
                        name = (await var.read_browse_name()).Name
                        val = await var.read_data_value()
                        self.add_variable(name, val.Value.Value, await var.read_data_type_as_variant_type())
                        for prop in await var.get_properties():
                            prop_name = (await prop.read_browse_name()).Name
                            name = '%s/%s' % (name, prop_name)
                            val = await prop.read_data_value()
                            self.add_property(name, val.Value.Value, val.Value.VariantType)
                    parents = await curr_node.get_referenced_nodes(refs=ua.ObjectIds.HasSubtype,
                                                                   direction=ua.BrowseDirection.Inverse,
                                                                   includesubtypes=True)
                    if len(parents) != 1:  # Something went wrong
                        raise UaError("Parent of event type could not be found")
                    curr_node = parents[0]
                self._freeze = True

        ce = CustomEvent()
        await ce.init()
        return ce


async def _find_parent_eventtype(node):
    """
    """
    parents = await node.get_referenced_nodes(refs=ua.ObjectIds.HasSubtype, direction=ua.BrowseDirection.Inverse,
                                              includesubtypes=True)

    if len(parents) != 1:   # Something went wrong
        raise UaError("Parent of event type could not be found")
    if parents[0].nodeid.NamespaceIndex == 0:
        if parents[0].nodeid.Identifier in asyncua.common.event_objects.IMPLEMENTED_EVENTS.keys():
            return parents[0].nodeid.Identifier, \
                   asyncua.common.event_objects.IMPLEMENTED_EVENTS[parents[0].nodeid.Identifier]
    return await _find_parent_eventtype(parents[0])


1		import copy
2
3		from asyncua import ua
4		import asyncua
5		from ..ua.uaerrors import UaError
6		from .ua_utils import get_node_subtypes
7
8
9		class Event:
10		"""
11		OPC UA Event object.
12		This is class in inherited by the common event objects such as BaseEvent,
13		other auto standard events and custom events
14		Events are used to trigger events on server side and are
15		sent to clients for every events from server
16
17		Developer Warning:
18		On server side the data type of attributes should be known, thus
19		add properties using the add_property method!!!
20		"""
21
22		def __init__(self, emitting_node=ua.ObjectIds.Server):
23		self.server_handle = None
24		self.select_clauses = None
25		self.event_fields = None
26		self.data_types = {}
27		self.emitting_node = emitting_node
28		if isinstance(emitting_node, ua.NodeId):
29		self.emitting_node = emitting_node
30		else:
31		self.emitting_node = ua.NodeId(emitting_node)
32		# save current attributes
33		self.internal_properties = list(self.__dict__.keys())[:] + ["internal_properties"]
34
35		def __str__(self):
36		return "{0}({1})".format(
37		self.__class__.__name__,
38		[str(k) + ":" + str(v) for k, v in self.__dict__.items() if k not in self.internal_properties])
39		__repr__ = __str__
40
41		def add_property(self, name, val, datatype):
42		"""
43		Add a property to event and store its data type
44		"""
45		setattr(self, name, val)
46		self.data_types[name] = datatype
47
48		def add_variable(self, name, val, datatype):
49		"""
50		Add a variable to event and store its data type
51		variables are able to have properties as children
52		"""
53		setattr(self, name, val)
54		self.data_types[name] = datatype
55
56		def get_event_props_as_fields_dict(self):
57		"""
58		convert all properties and variables of the Event class to a dict of variants
59		"""
60		field_vars = {}
61		for key, value in vars(self).items():
62		if not key.startswith("__") and key not in self.internal_properties:
63		field_vars[key] = ua.Variant(value, self.data_types[key])
64		return field_vars
65
66		@staticmethod
67		def from_field_dict(fields):
68		"""
69		Create an Event object from a dict of name and variants
70		"""
71		ev = Event()
72		for k, v in fields.items():
73		ev.add_property(k, v.Value, v.VariantType)
74		return ev
75
76		def to_event_fields_using_subscription_fields(self, select_clauses):
77		"""
78		Using a new select_clauses and the original select_clauses
79		used during subscription, return a field list
80		"""
81		fields = []
82		for sattr in select_clauses:
83		for idx, o_sattr in enumerate(self.select_clauses):
84		if sattr.BrowsePath == o_sattr.BrowsePath and sattr.AttributeId == o_sattr.AttributeId:
85		fields.append(self.event_fields[idx])
86		break
87		return fields
88
89		def to_event_fields(self, select_clauses):
90		"""
91		return a field list using a select clause and the object properties
92		"""
93		fields = []
94		for sattr in select_clauses:
95		if not sattr.BrowsePath:
96		name = ua.AttributeIds(sattr.AttributeId).name
97		else:
98		name = sattr.BrowsePath[0].Name
99		iter_paths = iter(sattr.BrowsePath)
100		next(iter_paths)
101		for path in iter_paths:
102		name += '/' + path.Name
103		try:
104		val = getattr(self, name)
105		except AttributeError:
106		field = ua.Variant(None)
107		else:
108		if val is None:
109		field = ua.Variant(None)
110		else:
111		field = ua.Variant(copy.deepcopy(val), self.data_types[name])
112		fields.append(field)
113		return fields
114
115		@staticmethod
116		def from_event_fields(select_clauses, fields):
117		"""
118		Instantiate an Event object from a select_clauses and fields
119		"""
120		ev = Event()
121		ev.select_clauses = select_clauses
122		ev.event_fields = fields
123		for idx, sattr in enumerate(select_clauses):
124		if len(sattr.BrowsePath) == 0:
125		name = sattr.AttributeId.name
126		else:
127		name = sattr.BrowsePath[0].Name
128		iter_paths = iter(sattr.BrowsePath)
129		next(iter_paths)
130		for path in iter_paths:
131		name += '/' + path.Name
132		ev.add_property(name, fields[idx].Value, fields[idx].VariantType)
133		return ev
134
135
136		async def get_filter_from_event_type(eventtypes):
137		evfilter = ua.EventFilter()
138		evfilter.SelectClauses = await select_clauses_from_evtype(eventtypes)
139		evfilter.WhereClause = await where_clause_from_evtype(eventtypes)
140		return evfilter
141
142
143		async def select_clauses_from_evtype(evtypes):
144		clauses = []
145		selected_paths = []
146		for evtype in evtypes:
147		event_props_and_vars = await get_event_properties_from_type_node(evtype)
148		event_props_and_vars.extend(await get_event_variables_from_type_node(evtype))
149		for node in event_props_and_vars:
150		browse_name = await node.read_browse_name()
151		if browse_name not in selected_paths:
152		op = ua.SimpleAttributeOperand()
153		op.AttributeId = ua.AttributeIds.Value
154		op.BrowsePath = [browse_name]
155		clauses.append(op)
156		selected_paths.append(browse_name)
157		for var in await get_event_variables_from_type_node(evtype):
158		browse_name = await var.get_browse_name()
159		if browse_name not in selected_paths:
160		op = ua.SimpleAttributeOperand()
161		op.AttributeId = ua.AttributeIds.Value
162		op.BrowsePath = [browse_name]
163		clauses.append(op)
164		selected_paths.append(browse_name)
165		for prop in await var.get_properties():
166		browse_path = [browse_name, await prop.get_browse_name()]
167		if browse_path not in selected_paths:
168		op = ua.SimpleAttributeOperand()
169		op.AttributeId = ua.AttributeIds.Value
170		op.BrowsePath = browse_path
171		clauses.append(op)
172		selected_paths.append(browse_path)
173		return clauses
174
175
176		async def where_clause_from_evtype(evtypes):
177		cf = ua.ContentFilter()
178		el = ua.ContentFilterElement()
179		# operands can be ElementOperand, LiteralOperand, AttributeOperand, SimpleAttribute
180		# Create a clause where the generate event type property EventType
181		# must be a subtype of events in evtypes argument
182
183		# the first operand is the attribute event type
184		op = ua.SimpleAttributeOperand()
185		# op.TypeDefinitionId = evtype.nodeid
186		op.BrowsePath.append(ua.QualifiedName("EventType", 0))
187		op.AttributeId = ua.AttributeIds.Value
188		el.FilterOperands.append(op)
189		# now create a list of all subtypes we want to accept
190		subtypes = []
191		for evtype in evtypes:
192		for st in await get_node_subtypes(evtype):
193		subtypes.append(st.nodeid)
194		subtypes = list(set(subtypes)) # remove duplicates
195		for subtypeid in subtypes:
196		op = ua.LiteralOperand()
197		op.Value = ua.Variant(subtypeid)
198		el.FilterOperands.append(op)
199		el.FilterOperator = ua.FilterOperator.InList
200		cf.Elements.append(el)
201		return cf
202
203
204	View Code Duplication	async def get_event_variables_from_type_node(node):
		0 ignored issues – show Duplication introduced 2020-12-18 10:17 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
205		variables = []
206		curr_node = node
207		while True:
208		variables.extend(await curr_node.get_variables())
209		if curr_node.nodeid.Identifier == ua.ObjectIds.BaseEventType:
210		break
211		parents = await curr_node.get_referenced_nodes(
212		refs=ua.ObjectIds.HasSubtype, direction=ua.BrowseDirection.Inverse, includesubtypes=True)
213		if len(parents) != 1: # Something went wrong
214		return None
215		curr_node = parents[0]
216		return variables
217
218
219		async def get_event_properties_from_type_node(node):
220		properties = []
221		curr_node = node
222		while True:
223		properties.extend(await curr_node.get_properties())
224		if curr_node.nodeid.Identifier == ua.ObjectIds.BaseEventType:
225		break
226		parents = await curr_node.get_referenced_nodes(
227		refs=ua.ObjectIds.HasSubtype, direction=ua.BrowseDirection.Inverse, includesubtypes=True
228		)
229		if len(parents) != 1: # Something went wrong
230		return None
231		curr_node = parents[0]
232		return properties
233
234
235	View Code Duplication	async def get_event_variables_from_type_node(node):
		0 ignored issues – show Duplication introduced 2020-12-18 10:17 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
236		variables = []
237		curr_node = node
238		while True:
239		variables.extend(await curr_node.get_variables())
240		if curr_node.nodeid.Identifier == ua.ObjectIds.BaseEventType:
241		break
242		parents = await curr_node.get_referenced_nodes(
243		refs=ua.ObjectIds.HasSubtype, direction=ua.BrowseDirection.Inverse, includesubtypes=True
244		)
245		if len(parents) != 1: # Something went wrong
246		return None
247		curr_node = parents[0]
248		return variables
249
250
251		async def get_event_obj_from_type_node(node):
252		"""
253		return an Event object from an event type node
254		"""
255		if node.nodeid.NamespaceIndex == 0 and \
256		node.nodeid.Identifier in asyncua.common.event_objects.IMPLEMENTED_EVENTS.keys():
257		return asyncua.common.event_objects.IMPLEMENTED_EVENTS[node.nodeid.Identifier]()
258		else:
259		parent_identifier, parent_eventtype = await _find_parent_eventtype(node)
260
261		class CustomEvent(parent_eventtype):
262
263		def __init__(self):
264		parent_eventtype.__init__(self)
		0 ignored issues – show introduced 2019-03-15 08:18 UTC by Report Bug Copy Issue Report The variable `parent_eventtype` does not seem to be defined for all execution paths. Loading history...
265		self.EventType = node.nodeid
266
267		async def init(self):
268		curr_node = node
269		while curr_node.nodeid.Identifier != parent_identifier:
		0 ignored issues – show introduced 2019-03-15 08:18 UTC by Report Bug Copy Issue Report The variable `parent_identifier` does not seem to be defined for all execution paths. Loading history...
270		for prop in await curr_node.get_properties():
271		name = (await prop.read_browse_name()).Name
272		val = await prop.read_data_value()
273		self.add_property(name, val.Value.Value, val.Value.VariantType)
274		for var in await curr_node.get_variables():
275		name = (await var.read_browse_name()).Name
276		val = await var.read_data_value()
277		self.add_variable(name, val.Value.Value, await var.read_data_type_as_variant_type())
278		for prop in await var.get_properties():
279		prop_name = (await prop.read_browse_name()).Name
280		name = '%s/%s' % (name, prop_name)
281		val = await prop.read_data_value()
282		self.add_property(name, val.Value.Value, val.Value.VariantType)
283		parents = await curr_node.get_referenced_nodes(refs=ua.ObjectIds.HasSubtype,
284		direction=ua.BrowseDirection.Inverse,
285		includesubtypes=True)
286		if len(parents) != 1: # Something went wrong
287		raise UaError("Parent of event type could not be found")
288		curr_node = parents[0]
289		self._freeze = True
290
291		ce = CustomEvent()
292		await ce.init()
293		return ce
294
295
296		async def _find_parent_eventtype(node):
297		"""
298		"""
299		parents = await node.get_referenced_nodes(refs=ua.ObjectIds.HasSubtype, direction=ua.BrowseDirection.Inverse,
300		includesubtypes=True)
301
302		if len(parents) != 1: # Something went wrong
303		raise UaError("Parent of event type could not be found")
304		if parents[0].nodeid.NamespaceIndex == 0:
305		if parents[0].nodeid.Identifier in asyncua.common.event_objects.IMPLEMENTED_EVENTS.keys():
306		return parents[0].nodeid.Identifier, \
307		asyncua.common.event_objects.IMPLEMENTED_EVENTS[parents[0].nodeid.Identifier]
308		return await _find_parent_eventtype(parents[0])
309

FreeOpcUa / opcua-asyncio

Pull Request — master (#365)

asyncua.common.events.Event.from_field_dict() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like