asyncua.common.events.Event.to_event_fields() - Code Metrics - Inspection of "Started to implement Alarms and Conditions" - FreeOpcUa/opcua-asyncio - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#101)

unknown

created 2019-11-26 11:16 UTC

asyncua.common.events.Event.to_event_fields() B

↳ Parent: asyncua.common.events

Complexity

Conditions

Size

Total Lines	25
Code Lines	19

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	7
eloc	19
nop	2
dl	0
loc	25
rs	8
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

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

    def __init__(self):
        self.server_handle = None
        self.select_clauses = None
        self.event_fields = None
        self.data_types = {}
        # 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 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:
        for prop in await get_event_properties_from_type_node(evtype):
            browse_name = await prop.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 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_obj_from_type_node(node):
    """
    return an Event object from an event type node
    """
    if node.nodeid.NamespaceIndex == 0:
        if node.nodeid.Identifier in asyncua.common.event_objects.IMPLEMENTED_EVENTS.keys():
            return asyncua.common.event_objects.IMPLEMENTED_EVENTS[node.nodeid.Identifier]()
    else:
        # for custom events or conditions
        node_tupple = (node.nodeid.NamespaceIndex, node.nodeid.Identifier)  # url as nsidx
        if node_tupple in asyncua.common.event_objects.IMPLEMENTED_EVENTS.keys():
            return asyncua.common.event_objects.IMPLEMENTED_EVENTS[node_tupple]()

    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.get_browse_name()).Name
                    val = await prop.get_data_value()
                    self.add_property(name, val.Value.Value, val.Value.VariantType)
                for var in await curr_node.get_variables():
                    name = (await var.get_browse_name()).Name
                    val = await var.get_data_value()
                    self.add_variable(name, val.Value.Value, var.get_data_type_as_variant_type())
                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 notbe 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 notbe found")
    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]
    else:
        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			Developper 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):
23			self.server_handle = None
24			self.select_clauses = None
25			self.event_fields = None
26			self.data_types = {}
27			# save current attributes
28			self.internal_properties = list(self.__dict__.keys())[:] + ["internal_properties"]
29
30			def __str__(self):
31			return "{0}({1})".format(
32			self.__class__.__name__,
33			[str(k) + ":" + str(v) for k, v in self.__dict__.items() if k not in self.internal_properties])
34			__repr__ = __str__
35
36			def add_property(self, name, val, datatype):
37			"""
38			Add a property to event and store its data type
39			"""
40			setattr(self, name, val)
41			self.data_types[name] = datatype
42
43			def add_variable(self, name, val, datatype):
44			"""
45			Add a variable to event and store its data type
46			variables are able to have children
47			"""
48			setattr(self, name, val)
49			self.data_types[name] = datatype
50
51			def get_event_props_as_fields_dict(self):
52			"""
53			convert all properties and variables of the Event class to a dict of variants
54			"""
55			field_vars = {}
56			for key, value in vars(self).items():
57			if not key.startswith("__") and key not in self.internal_properties:
58			field_vars[key] = ua.Variant(value, self.data_types[key])
59			return field_vars
60
61			@staticmethod
62			def from_field_dict(fields):
63			"""
64			Create an Event object from a dict of name and variants
65			"""
66			ev = Event()
67			for k, v in fields.items():
68			ev.add_property(k, v.Value, v.VariantType)
69			return ev
70
71			def to_event_fields_using_subscription_fields(self, select_clauses):
72			"""
73			Using a new select_clauses and the original select_clauses
74			used during subscription, return a field list
75			"""
76			fields = []
77			for sattr in select_clauses:
78			for idx, o_sattr in enumerate(self.select_clauses):
79			if sattr.BrowsePath == o_sattr.BrowsePath and sattr.AttributeId == o_sattr.AttributeId:
80			fields.append(self.event_fields[idx])
81			break
82			return fields
83
84			def to_event_fields(self, select_clauses):
85			"""
86			return a field list using a select clause and the object properties
87			"""
88			fields = []
89			for sattr in select_clauses:
90			if not sattr.BrowsePath:
91			name = ua.AttributeIds(sattr.AttributeId).name
92			else:
93			name = sattr.BrowsePath[0].Name
94			iter_paths = iter(sattr.BrowsePath)
95			next(iter_paths)
96			for path in iter_paths:
97			name += '/' + path.Name
98			try:
99			val = getattr(self, name)
100			except AttributeError:
101			field = ua.Variant(None)
102			else:
103			if val is None:
104			field = ua.Variant(None)
105			else:
106			field = ua.Variant(copy.deepcopy(val), self.data_types[name])
107			fields.append(field)
108			return fields
109
110			@staticmethod
111			def from_event_fields(select_clauses, fields):
112			"""
113			Instantiate an Event object from a select_clauses and fields
114			"""
115			ev = Event()
116			ev.select_clauses = select_clauses
117			ev.event_fields = fields
118			for idx, sattr in enumerate(select_clauses):
119			if len(sattr.BrowsePath) == 0:
120			name = sattr.AttributeId.name
121			else:
122			name = sattr.BrowsePath[0].Name
123			iter_paths = iter(sattr.BrowsePath)
124			next(iter_paths)
125			for path in iter_paths:
126			name += '/' + path.Name
127			ev.add_property(name, fields[idx].Value, fields[idx].VariantType)
128			return ev
129
130
131			async def get_filter_from_event_type(eventtypes):
132			evfilter = ua.EventFilter()
133			evfilter.SelectClauses = await select_clauses_from_evtype(eventtypes)
134			evfilter.WhereClause = await where_clause_from_evtype(eventtypes)
135			return evfilter
136
137
138			async def select_clauses_from_evtype(evtypes):
139			clauses = []
140			selected_paths = []
141			for evtype in evtypes:
142			for prop in await get_event_properties_from_type_node(evtype):
143			browse_name = await prop.get_browse_name()
144			if browse_name not in selected_paths:
145			op = ua.SimpleAttributeOperand()
146			op.AttributeId = ua.AttributeIds.Value
147			op.BrowsePath = [browse_name]
148			clauses.append(op)
149			selected_paths.append(browse_name)
150			for var in await get_event_variables_from_type_node(evtype):
151			browse_name = await var.get_browse_name()
152			if browse_name not in selected_paths:
153			op = ua.SimpleAttributeOperand()
154			op.AttributeId = ua.AttributeIds.Value
155			op.BrowsePath = [browse_name]
156			clauses.append(op)
157			selected_paths.append(browse_name)
158			for prop in await var.get_properties():
159			browse_path = [browse_name, await prop.get_browse_name()]
160			if browse_path not in selected_paths:
161			op = ua.SimpleAttributeOperand()
162			op.AttributeId = ua.AttributeIds.Value
163			op.BrowsePath = browse_path
164			clauses.append(op)
165			selected_paths.append(browse_path)
166			return clauses
167
168
169			async def where_clause_from_evtype(evtypes):
170			cf = ua.ContentFilter()
171			el = ua.ContentFilterElement()
172			# operands can be ElementOperand, LiteralOperand, AttributeOperand, SimpleAttribute
173			# Create a clause where the generate event type property EventType
174			# must be a subtype of events in evtypes argument
175
176			# the first operand is the attribute event type
177			op = ua.SimpleAttributeOperand()
178			# op.TypeDefinitionId = evtype.nodeid
179			op.BrowsePath.append(ua.QualifiedName("EventType", 0))
180			op.AttributeId = ua.AttributeIds.Value
181			el.FilterOperands.append(op)
182			# now create a list of all subtypes we want to accept
183			subtypes = []
184			for evtype in evtypes:
185			for st in await get_node_subtypes(evtype):
186			subtypes.append(st.nodeid)
187			subtypes = list(set(subtypes)) # remove duplicates
188			for subtypeid in subtypes:
189			op = ua.LiteralOperand()
190			op.Value = ua.Variant(subtypeid)
191			el.FilterOperands.append(op)
192			el.FilterOperator = ua.FilterOperator.InList
193			cf.Elements.append(el)
194			return cf
195
196
197			async def get_event_variables_from_type_node(node):
198			variables = []
199			curr_node = node
200			while True:
201			variables.extend(await curr_node.get_variables())
202			if curr_node.nodeid.Identifier == ua.ObjectIds.BaseEventType:
203			break
204			parents = await curr_node.get_referenced_nodes(
205			refs=ua.ObjectIds.HasSubtype, direction=ua.BrowseDirection.Inverse, includesubtypes=True)
206			if len(parents) != 1: # Something went wrong
207			return None
208			curr_node = parents[0]
209			return variables
210
211
212			async def get_event_properties_from_type_node(node):
213			properties = []
214			curr_node = node
215			while True:
216			properties.extend(await curr_node.get_properties())
217			if curr_node.nodeid.Identifier == ua.ObjectIds.BaseEventType:
218			break
219			parents = await curr_node.get_referenced_nodes(
220			refs=ua.ObjectIds.HasSubtype, direction=ua.BrowseDirection.Inverse, includesubtypes=True
221			)
222			if len(parents) != 1: # Something went wrong
223			return None
224			curr_node = parents[0]
225			return properties
226
227
228			async def get_event_obj_from_type_node(node):
229			"""
230			return an Event object from an event type node
231			"""
232			if node.nodeid.NamespaceIndex == 0:
233			if node.nodeid.Identifier in asyncua.common.event_objects.IMPLEMENTED_EVENTS.keys():
234			return asyncua.common.event_objects.IMPLEMENTED_EVENTS[node.nodeid.Identifier]()
235			else:
236			# for custom events or conditions
237			node_tupple = (node.nodeid.NamespaceIndex, node.nodeid.Identifier) # url as nsidx
238			if node_tupple in asyncua.common.event_objects.IMPLEMENTED_EVENTS.keys():
239			return asyncua.common.event_objects.IMPLEMENTED_EVENTS[node_tupple]()
240
241			parent_identifier, parent_eventtype = await _find_parent_eventtype(node)
242
243			class CustomEvent(parent_eventtype):
244
245			def __init__(self):
246			parent_eventtype.__init__(self)
247			self.EventType = node.nodeid
248
249			async def init(self):
250			curr_node = node
251			while curr_node.nodeid.Identifier != parent_identifier:
252			for prop in await curr_node.get_properties():
253			name = (await prop.get_browse_name()).Name
254			val = await prop.get_data_value()
255			self.add_property(name, val.Value.Value, val.Value.VariantType)
256			for var in await curr_node.get_variables():
257			name = (await var.get_browse_name()).Name
258			val = await var.get_data_value()
259			self.add_variable(name, val.Value.Value, var.get_data_type_as_variant_type())
260			parents = await curr_node.get_referenced_nodes(refs=ua.ObjectIds.HasSubtype,
261			direction=ua.BrowseDirection.Inverse,
262			includesubtypes=True)
263
264			if len(parents) != 1: # Something went wrong
265			raise UaError("Parent of event type could notbe found")
266			curr_node = parents[0]
267
268			self._freeze = True
269
270			ce = CustomEvent()
271			await ce.init()
272			return ce
273
274
275			async def _find_parent_eventtype(node):
276			"""
277			"""
278			parents = await node.get_referenced_nodes(refs=ua.ObjectIds.HasSubtype, direction=ua.BrowseDirection.Inverse, includesubtypes=True)
279
280			if len(parents) != 1: # Something went wrong
281			raise UaError("Parent of event type could notbe found")
282			if parents[0].nodeid.Identifier in asyncua.common.event_objects.IMPLEMENTED_EVENTS.keys():
283			return parents[0].nodeid.Identifier, asyncua.common.event_objects.IMPLEMENTED_EVENTS[parents[0].nodeid.Identifier]
284			else:
285			return await _find_parent_eventtype(parents[0])
286

FreeOpcUa / opcua-asyncio

Pull Request — master (#101)

asyncua.common.events.Event.to_event_fields() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like