get_event_from_node()

Complexity

Conditions

3

Size

Total Lines

46

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	12
CRAP Score	5.0721

Importance

Changes	1
Bugs	0	Features	0

2 Methods

Rating	Name	Duplication	Size	Complexity
A	CustomEvent.__str__()	0	12	1
A	CustomEvent.__init__()	0	12	4

from datetime import datetime

from opcua import ua
from opcua import Node
import uuid


class EventGenerator(object):

    """
    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

        source: 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
    """

    def __init__(self, isession, etype=ua.BaseEvent(), source=ua.ObjectIds.Server):
        self.isession = isession
        self.event = None
        node = None

        if isinstance(etype, ua.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 = get_event_from_node(node)

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

        if self.event.SourceNode.Identifier:
            source = Node(self.isession, self.event.SourceNode)

        self.event.SourceName = source.get_display_name().Text
        source.set_attribute(ua.AttributeIds.EventNotifier, ua.DataValue(ua.Variant(1, ua.VariantType.Byte)))

    def __str__(self):
        return "EventGenerator(Type:{}, Source:{}, Time:{}, Message: {})".format(self.EventType, self.SourceNode, self.Time, self.Message)
    __repr__ = __str__

    def trigger(self, time=None, message=None):
        """
        Trigger the event. This will send a notification to all subscribed clients
        """
        self.event.EventId = ua.Variant(uuid.uuid4().hex, ua.VariantType.ByteString)
        if time:
            self.event.Time = time
        else:
            self.event.Time = datetime.utcnow()
        self.event.RecieveTime = datetime.utcnow()
        #FIXME: LocalTime is wrong but currently know better. For description s. Part 5 page 18
        self.event.LocalTime = datetime.utcnow()
        if message:
            self.Message = ua.LocalizedText(message)
        elif not self.event.Message:
            self.event.Message = ua.LocalizedText(self.source.get_browse_name().Text)
        self.isession.subscription_service.trigger_event(self.event)


def get_event_from_node(node):
    if node.nodeid.Identifier in ua.uaevents_auto.IMPLEMNTED_EVENTS.keys():
        return ua.uaevents_auto.IMPLEMNTED_EVENTS[node.nodeid.Identifier]()
    else:
        parent_identifier, parent_eventtype = _find_parent_eventtype(node)
        if not parent_eventtype:
            return None

        class CustomEvent(parent_eventtype):

            def __init__(self):
                super(CustomEvent, self).__init__()
                curr_node = node
                while curr_node.nodeid.Identifier != parent_identifier:
                    properties = curr_node.get_referenced_nodes(refs=ua.ObjectIds.HasProperty, direction=ua.BrowseDirection.Forward)
                    for prop in properties:
                        setattr(self, prop.get_browse_name().Name, prop.get_value())

                    parents = node.get_referenced_nodes(refs=ua.ObjectIds.HasSubtype, direction=ua.BrowseDirection.Inverse, includesubtypes=False)
                    if len(parents) != 1: # Something went wrong
                        return None
                    curr_node = parents[0]

            #TODO: Extend to show all fields of CustomEvent
            def __str__(self):
                s = 'CustomEvent('
                s += 'EventId:{}'.format(self.EventId)
                s += ', EventType:{}'.format(self.EventType)
                s += ', SourceNode:{}'.format(self.SourceNode)
                s += ', SourceName:{}'.format(self.SourceName)
                s += ', Time:{}'.format(self.Time)
                s += ', RecieveTime:{}'.format(self.RecieveTime)
                s += ', LocalTime:{}'.format(self.LocalTime)
                s += ', Message:{}'.format(self.Message)
                s += ', Severity:{}'.format(self.Severity)
                s += ')'

        #class CustomEvent():

            #pass
    #references = node.get_children_descriptions(refs=ua.ObjectIds.HasProperty)
    #for desc in references:
        #child = Node(self.isession, desc.NodeId)
        #setattr(self.event, desc.BrowseName.Name, child.get_value())

    return CustomEvent()


def _find_parent_eventtype(node):
    parents = node.get_referenced_nodes(refs=ua.ObjectIds.HasSubtype, direction=ua.BrowseDirection.Inverse, includesubtypes=False)

    if len(parents) != 1:   # Something went wrong
        return None
    if parents[0].nodeid.Identifier in ua.uaevents_auto.IMPLEMNTED_EVENTS.keys():
        return node.nodeid.Identifier, ua.uaevents_auto.IMPLEMNTED_EVENTS[parents[0].nodeid.Identifier]
    else:
        _find_parent_eventtype(parents[0])