small renaming, mv more methods to events.py

opcua/__init__.py

@@ -13,7 +13,7 @@ from opcua.common.methods import uamethod
 from opcua.common.subscription import Subscription
 from opcua.client.client import Client
 from opcua.server.server import Server
-from opcua.server.event import EventGenerator
+from opcua.server.event_generator import EventGenerator
 from opcua.common.instanciate import instanciate_node
 
 

opcua/common/events.py

@@ -148,3 +148,46 @@ def get_event_properties_from_type_node(node):
     return properties
 
 
+def get_event_obj_from_type_node(node):
+    """
+    return an Event object from an event type node
+    """
+    if node.nodeid.Identifier in ua.uaevents_auto.IMPLEMENTED_EVENTS.keys():
+        return ua.uaevents_auto.IMPLEMENTED_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__(extended=True)
+                self.EventType = node.nodeid
+                curr_node = node
+
+                while curr_node.nodeid.Identifier != parent_identifier:
+                    for prop in curr_node.get_properties():
+                        setattr(self, prop.get_browse_name().Name, prop.get_value())
+                    parents = curr_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]
+
+                self._freeze = True
+
+    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, None
+    if parents[0].nodeid.Identifier in ua.uaevents_auto.IMPLEMENTED_EVENTS.keys():
+        return parents[0].nodeid.Identifier, ua.uaevents_auto.IMPLEMENTED_EVENTS[parents[0].nodeid.Identifier]
+    else:
+        return _find_parent_eventtype(parents[0])
+

opcua/server/event.py → opcua/server/event_generator.py

 import logging
 from datetime import datetime
+import uuid
 
 from opcua import ua
 from opcua import Node
-import uuid
+from opcua.common import events
 
 
 class EventGenerator(object):
@@ -41,7 +42,7 @@ class EventGenerator(object):
             node = Node(self.isession, ua.NodeId(etype))
 
         if node:
-            self.event = get_event_from_type_node(node)
+            self.event = events.get_event_obj_from_type_node(node)
 
         if isinstance(source, Node):
             pass
@@ -52,7 +53,8 @@ class EventGenerator(object):
 
         if self.event.SourceNode:
             if source.nodeid != self.event.SourceNode:
-                self.logger.warning("Source NodeId: '%s' and event SourceNode: '%s' are not the same. Using '%s' as SourceNode", str(source.nodeid), str(self.event.SourceNode), str(self.event.SourceNode))
+                self.logger.warning(
+                    "Source NodeId: '%s' and event SourceNode: '%s' are not the same. Using '%s' as SourceNode", str(source.nodeid), str(self.event.SourceNode), str(self.event.SourceNode))
                 source = Node(self.isession, self.event.SourceNode)
 
         self.event.SourceNode = source.nodeid
@@ -96,40 +98,3 @@ class EventGenerator(object):
         self.isession.subscription_service.trigger_event(self.event)
 
 
-def get_event_from_type_node(node):
-    if node.nodeid.Identifier in ua.uaevents_auto.IMPLEMENTED_EVENTS.keys():
-        return ua.uaevents_auto.IMPLEMENTED_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__(extended=True)
-                self.EventType = node.nodeid
-                curr_node = node
-
-                while curr_node.nodeid.Identifier != parent_identifier:
-                    for prop in curr_node.get_properties():
-                        setattr(self, prop.get_browse_name().Name, prop.get_value())
-                    parents = curr_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]
-
-                self._freeze = True
-
-    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, None
-    if parents[0].nodeid.Identifier in ua.uaevents_auto.IMPLEMENTED_EVENTS.keys():
-        return parents[0].nodeid.Identifier, ua.uaevents_auto.IMPLEMENTED_EVENTS[parents[0].nodeid.Identifier]
-    else:
-        return _find_parent_eventtype(parents[0])

opcua/server/server.py

@@ -13,7 +13,7 @@ from opcua import ua
 #from opcua.binary_server import BinaryServer
 from opcua.server.binary_server_asyncio import BinaryServer
 from opcua.server.internal_server import InternalServer
-from opcua.server.event import EventGenerator
+from opcua.server.event_generator import EventGenerator
 from opcua.common.node import Node
 from opcua.common.subscription import Subscription
 from opcua.common import xmlimporter

tests/tests_server.py

@@ -173,11 +173,11 @@ class TestServer(unittest.TestCase, CommonTests, SubscriptionTests):
 
     # This should work for following BaseEvent tests to work (maybe to write it a bit differentlly since they are not independent)
     def test_get_event_from_type_node_BaseEvent(self):
-        ev = opcua.server.event.get_event_from_type_node(opcua.Node(self.opc.iserver.isession, ua.NodeId(ua.ObjectIds.BaseEventType)))
+        ev = opcua.common.events.get_event_obj_from_type_node(opcua.Node(self.opc.iserver.isession, ua.NodeId(ua.ObjectIds.BaseEventType)))
         check_base_event(self, ev)
 
     def test_get_event_from_type_node_Inhereted_AuditEvent(self):
-        ev = opcua.server.event.get_event_from_type_node(opcua.Node(self.opc.iserver.isession, ua.NodeId(ua.ObjectIds.AuditEventType)))
+        ev = opcua.common.events.get_event_obj_from_type_node(opcua.Node(self.opc.iserver.isession, ua.NodeId(ua.ObjectIds.AuditEventType)))
         self.assertIsNot(ev, None)  # we did not receive event
         self.assertIsInstance(ev, ua.BaseEvent)
         self.assertIsInstance(ev, ua.AuditEvent)
@@ -282,7 +282,7 @@ class TestServer(unittest.TestCase, CommonTests, SubscriptionTests):
     def test_get_event_from_type_node_CustomEvent(self):
         etype = self.opc.create_custom_event_type(2, 'MyEvent', ua.ObjectIds.BaseEventType, [('PropertyNum', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)])
 
-        ev = opcua.server.event.get_event_from_type_node(etype)
+        ev = opcua.common.events.get_event_obj_from_type_node(etype)
         check_custom_event(self, ev, etype)
         self.assertEqual(ev.PropertyNum, None)
         self.assertEqual(ev.PropertyString, None)