asyncua.common.events.Event.__init__() - 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-20 14:46 UTC

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

↳ Parent: asyncua.common.events

Complexity

Conditions

Size

Total Lines	8
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

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

FreeOpcUa / opcua-asyncio

Pull Request — master (#101)

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

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

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