asyncua.server.event_generator.EventGenerator.init() - 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-22 09:16 UTC

EventGenerator.init() F

↳ Parent: asyncua.server.event_generator

Complexity

Conditions

Size

Total Lines	83
Code Lines	70

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	22
eloc	70
nop	4
dl	0
loc	83
rs	0
c	0
b	0
f	0

How to fix Long Method Complexity

import logging
from datetime import datetime
import time
import uuid
from typing import Optional
import sys

from asyncua import ua
from asyncua.ua.uaerrors import BadSourceNodeIdInvalid, BadTargetNodeIdInvalid
from ..common import events, event_objects, Node


class EventGenerator:
    """
    Create an event based on an event type. Per default is BaseEventType used.
    Object members are dynamically created from the base event type and send to
    client when evebt is triggered (see example code in source)

    Arguments to constructor are:
        server: The InternalSession object to use for query and event triggering
        emitting_node: The emiting source for the node, either an objectId, NodeId or a Node
        etype: The event type, either an objectId, a NodeId or a Node object
        notifier_path: the path from the Server-Object to the emitting node
    """

    def __init__(self, isession):
        self.logger = logging.getLogger(__name__)
        self.isession = isession
        self.event: Optional[event_objects.BaseEvent] = None
        self.emitting_node: Optional[Node] = None

    async def init(self, etype=None, emitting_node=ua.ObjectIds.Server, notifier_path=None):
        node = None

        if isinstance(etype, event_objects.BaseEvent):
            self.event = etype
        elif isinstance(etype, Node):
            node = etype
        elif isinstance(etype, ua.NodeId):
            node = Node(self.isession, etype)
        else:
            node = Node(self.isession, ua.NodeId(etype))

        if node:
            self.event = await events.get_event_obj_from_type_node(node)
            if isinstance(self.event, event_objects.Condition):
                # we need this that it works proper for conditions and alarms aka ConditionId
                if isinstance(emitting_node, Node):
                    condition_id = ua.NodeId(emitting_node.nodeid.Identifier, emitting_node.nodeid.NamespaceIndex)
                elif isinstance(emitting_node, ua.NodeId):
                    condition_id = emitting_node
                elif isinstance(emitting_node, int):
                    condition_id = ua.NodeId(emitting_node)
                else:
                    condition_id = ua.NodeId(emitting_node.Identifier, emitting_node.NamespaceIndex)
                self.event.add_property('NodeId', condition_id, ua.VariantType.NodeId)

        if isinstance(emitting_node, Node):
            pass
        elif isinstance(emitting_node, ua.NodeId):
            emitting_node = Node(self.isession, emitting_node)
        else:
            emitting_node = Node(self.isession, ua.NodeId(emitting_node))

        self.event.emitting_node = emitting_node.nodeid
        if not self.event.SourceNode:
            self.event.SourceNode = emitting_node.nodeid
        if not self.event.SourceName:
            self.event.SourceName = (await Node(self.isession, self.event.SourceNode).read_browse_name()).Name

        refs = []
        if notifier_path is not None:
            for i, node in enumerate(notifier_path):
                if not isinstance(node, Node):
                    notifier_path[i] = Node(self.isession, node)
            if notifier_path[0] != Node(self.isession, ua.ObjectIds.Server):
                raise BadSourceNodeIdInvalid
            if notifier_path[-1] != emitting_node:
                raise BadTargetNodeIdInvalid
            for i, node in enumerate(notifier_path):
                if node == emitting_node:
                    continue
                ref = ua.AddReferencesItem()
                ref.IsForward = True
                ref.ReferenceTypeId = ua.NodeId(ua.ObjectIds.HasNotifier)
                ref.SourceNodeId = node.nodeid
                ref.TargetNodeClass = await node.read_node_class()
                ref.TargetNodeId = notifier_path[i + 1].nodeid
                refs.append(ref)

                await node.set_event_notifier([ua.EventNotifier.SubscribeToEvents])
        else:
            if emitting_node.nodeid.Identifier != ua.ObjectIds.Server:
                ref = ua.AddReferencesItem()
                ref.IsForward = True
                ref.ReferenceTypeId = ua.NodeId(ua.ObjectIds.HasNotifier)
                ref.SourceNodeId = ua.NodeId(ua.ObjectIds.Server)
                ref.TargetNodeClass = await emitting_node.read_node_class()
                ref.TargetNodeId = emitting_node.nodeid
                refs.append(ref)

        await emitting_node.set_event_notifier([ua.EventNotifier.SubscribeToEvents])
        ref = ua.AddReferencesItem()
        ref.IsForward = True
        ref.ReferenceTypeId = ua.NodeId(ua.ObjectIds.GeneratesEvent)
        ref.SourceNodeId = emitting_node.nodeid
        ref.TargetNodeClass = ua.NodeClass.ObjectType
        ref.TargetNodeId = self.event.EventType
        refs.append(ref)
        results = await self.isession.add_references(refs)
        for result in results:
            result.check()

        self.emitting_node = emitting_node

    def __str__(self):
        return f"EventGenerator(Type:{self.event.EventType}, Emitting Node:{self.emitting_node}, " \
               f"Time:{self.event.Time}, Message: {self.event.Message})"

    __repr__ = __str__

    async def trigger(self, time_attr=None, message=None):
        """
        Trigger the event. This will send a notification to all subscribed clients
        """
        self.event.EventId = ua.Variant(uuid.uuid4().hex.encode('utf-8'), ua.VariantType.ByteString)
        if time_attr:
            self.event.Time = time_attr
        else:
            self.event.Time = datetime.utcnow()
        self.event.ReceiveTime = datetime.utcnow()

        self.event.LocalTime = ua.uaprotocol_auto.TimeZoneDataType()
        if sys.version_info.major > 2:
            localtime = time.localtime(self.event.Time.timestamp())
            self.event.LocalTime.Offset = localtime.tm_gmtoff//60
        else:
            localtime = time.localtime(time.mktime(self.event.Time.timetuple()))
            self.event.LocalTime.Offset = -(time.altzone if localtime.tm_isdst else time.timezone)
        self.event.LocalTime.DaylightSavingInOffset = bool(localtime.tm_isdst != -1)

        if message:
            self.event.Message = ua.LocalizedText(message)
        elif not self.event.Message:
            self.event.Message = ua.LocalizedText((await Node(self.isession, self.event.SourceNode).read_browse_name()).Name).Text
        await self.isession.subscription_service.trigger_event(self.event)


1			import logging
2			from datetime import datetime
3			import time
4			import uuid
5			from typing import Optional
6			import sys
7
8			from asyncua import ua
9			from asyncua.ua.uaerrors import BadSourceNodeIdInvalid, BadTargetNodeIdInvalid
10			from ..common import events, event_objects, Node
11
12
13			class EventGenerator:
14			"""
15			Create an event based on an event type. Per default is BaseEventType used.
16			Object members are dynamically created from the base event type and send to
17			client when evebt is triggered (see example code in source)
18
19			Arguments to constructor are:
20			server: The InternalSession object to use for query and event triggering
21			emitting_node: The emiting source for the node, either an objectId, NodeId or a Node
22			etype: The event type, either an objectId, a NodeId or a Node object
23			notifier_path: the path from the Server-Object to the emitting node
24			"""
25
26			def __init__(self, isession):
27			self.logger = logging.getLogger(__name__)
28			self.isession = isession
29			self.event: Optional[event_objects.BaseEvent] = None
30			self.emitting_node: Optional[Node] = None
31
32			async def init(self, etype=None, emitting_node=ua.ObjectIds.Server, notifier_path=None):
33			node = None
34
35			if isinstance(etype, event_objects.BaseEvent):
36			self.event = etype
37			elif isinstance(etype, Node):
38			node = etype
39			elif isinstance(etype, ua.NodeId):
40			node = Node(self.isession, etype)
41			else:
42			node = Node(self.isession, ua.NodeId(etype))
43
44			if node:
45			self.event = await events.get_event_obj_from_type_node(node)
46			if isinstance(self.event, event_objects.Condition):
47			# we need this that it works proper for conditions and alarms aka ConditionId
48			if isinstance(emitting_node, Node):
49			condition_id = ua.NodeId(emitting_node.nodeid.Identifier, emitting_node.nodeid.NamespaceIndex)
50			elif isinstance(emitting_node, ua.NodeId):
51			condition_id = emitting_node
52			elif isinstance(emitting_node, int):
53			condition_id = ua.NodeId(emitting_node)
54			else:
55			condition_id = ua.NodeId(emitting_node.Identifier, emitting_node.NamespaceIndex)
56			self.event.add_property('NodeId', condition_id, ua.VariantType.NodeId)
57
58			if isinstance(emitting_node, Node):
59			pass
60			elif isinstance(emitting_node, ua.NodeId):
61			emitting_node = Node(self.isession, emitting_node)
62			else:
63			emitting_node = Node(self.isession, ua.NodeId(emitting_node))
64
65			self.event.emitting_node = emitting_node.nodeid
66			if not self.event.SourceNode:
67			self.event.SourceNode = emitting_node.nodeid
68			if not self.event.SourceName:
69			self.event.SourceName = (await Node(self.isession, self.event.SourceNode).read_browse_name()).Name
70
71			refs = []
72			if notifier_path is not None:
73			for i, node in enumerate(notifier_path):
74			if not isinstance(node, Node):
75			notifier_path[i] = Node(self.isession, node)
76			if notifier_path[0] != Node(self.isession, ua.ObjectIds.Server):
77			raise BadSourceNodeIdInvalid
78			if notifier_path[-1] != emitting_node:
79			raise BadTargetNodeIdInvalid
80			for i, node in enumerate(notifier_path):
81			if node == emitting_node:
82			continue
83			ref = ua.AddReferencesItem()
84			ref.IsForward = True
85			ref.ReferenceTypeId = ua.NodeId(ua.ObjectIds.HasNotifier)
86			ref.SourceNodeId = node.nodeid
87			ref.TargetNodeClass = await node.read_node_class()
88			ref.TargetNodeId = notifier_path[i + 1].nodeid
89			refs.append(ref)
90
91			await node.set_event_notifier([ua.EventNotifier.SubscribeToEvents])
92			else:
93			if emitting_node.nodeid.Identifier != ua.ObjectIds.Server:
94			ref = ua.AddReferencesItem()
95			ref.IsForward = True
96			ref.ReferenceTypeId = ua.NodeId(ua.ObjectIds.HasNotifier)
97			ref.SourceNodeId = ua.NodeId(ua.ObjectIds.Server)
98			ref.TargetNodeClass = await emitting_node.read_node_class()
99			ref.TargetNodeId = emitting_node.nodeid
100			refs.append(ref)
101
102			await emitting_node.set_event_notifier([ua.EventNotifier.SubscribeToEvents])
103			ref = ua.AddReferencesItem()
104			ref.IsForward = True
105			ref.ReferenceTypeId = ua.NodeId(ua.ObjectIds.GeneratesEvent)
106			ref.SourceNodeId = emitting_node.nodeid
107			ref.TargetNodeClass = ua.NodeClass.ObjectType
108			ref.TargetNodeId = self.event.EventType
109			refs.append(ref)
110			results = await self.isession.add_references(refs)
111			for result in results:
112			result.check()
113
114			self.emitting_node = emitting_node
115
116			def __str__(self):
117			return f"EventGenerator(Type:{self.event.EventType}, Emitting Node:{self.emitting_node}, " \
118			f"Time:{self.event.Time}, Message: {self.event.Message})"
119
120			__repr__ = __str__
121
122			async def trigger(self, time_attr=None, message=None):
123			"""
124			Trigger the event. This will send a notification to all subscribed clients
125			"""
126			self.event.EventId = ua.Variant(uuid.uuid4().hex.encode('utf-8'), ua.VariantType.ByteString)
127			if time_attr:
128			self.event.Time = time_attr
129			else:
130			self.event.Time = datetime.utcnow()
131			self.event.ReceiveTime = datetime.utcnow()
132
133			self.event.LocalTime = ua.uaprotocol_auto.TimeZoneDataType()
134			if sys.version_info.major > 2:
135			localtime = time.localtime(self.event.Time.timestamp())
136			self.event.LocalTime.Offset = localtime.tm_gmtoff//60
137			else:
138			localtime = time.localtime(time.mktime(self.event.Time.timetuple()))
139			self.event.LocalTime.Offset = -(time.altzone if localtime.tm_isdst else time.timezone)
140			self.event.LocalTime.DaylightSavingInOffset = bool(localtime.tm_isdst != -1)
141
142			if message:
143			self.event.Message = ua.LocalizedText(message)
144			elif not self.event.Message:
145			self.event.Message = ua.LocalizedText((await Node(self.isession, self.event.SourceNode).read_browse_name()).Name).Text
146			await self.isession.subscription_service.trigger_event(self.event)
147

FreeOpcUa / opcua-asyncio

Pull Request — master (#365)

EventGenerator.init() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like