Custom type methods with examples

examples/complex-type-client-server/common.py

@@ -1,7 +1,7 @@
 from opcua.common.methods import to_variant
 from opcua.ua import ua_binary as uabin
 from opcua.ua.uatypes import Variant
-from opcua.ua.uautils import register_extension_object
+from opcua.common.ua_utils import register_extension_object
 
 
 class KeyValuePair(object):

opcua/common/ua_utils.py

@@ -9,6 +9,8 @@
 
 from opcua import ua
 from opcua.ua.uaerrors import UaError
+from opcua.ua import ObjectIds, ObjectIdNames
+from opcua.ua.uaprotocol_auto import ExtensionClasses
 
 
 def val_to_string(val):
@@ -256,3 +258,9 @@ def get_nodes_of_namespace(server, namespaces=[]):
     nodes = [server.get_node(nodeid) for nodeid in server.iserver.aspace.keys()
              if nodeid.NamespaceIndex != 0 and nodeid.NamespaceIndex in namespace_indexes]
     return nodes
+
+
+def register_extension_object(object_factory):
+    setattr(ObjectIds, "{}_Encoding_DefaultBinary".format(object_factory.__name__), object_factory.DEFAULT_BINARY)
+    ObjectIdNames[object_factory.DEFAULT_BINARY] = object_factory.__name__
+    ExtensionClasses[object_factory.DEFAULT_BINARY] = object_factory

opcua/ua/uatypes.py

@@ -629,6 +629,7 @@ def to_binary(self):
     def from_binary(data):
         obj = ExtensionObject()
         obj.TypeId = NodeId.from_binary(data)
+        obj.Encoding = uabin.Primitives.UInt8.unpack(data)
         if obj.Encoding & (1 << 0):
             obj.Body = uabin.Primitives.ByteString.unpack(data)
         return obj

tests/tests_common.py

@@ -7,11 +7,13 @@
 from opcua import uamethod
 from opcua import instantiate
 from opcua import copy_node
-from opcua.ua.uautils import register_extension_object
+from opcua.common.ua_utils import register_extension_object
 
 from opcua.ua import ua_binary as uabin
 from opcua.common import ua_utils
 
+import io
+
 
 def add_server_methods(srv):
     @uamethod
@@ -720,10 +722,10 @@ def unique_response(parent):
         self.assertEqual(type(response[0]), ua.Variant)
 
     def test_ua_method_complex(self):
-        register_extension_object(KeyValuePair)
+        register_extension_object(MyCustomClass)
 
         def unique_response(parent):
-            return KeyValuePair("Key", "Value")
+            return MyCustomClass("Key", "Value")
 
         decorated_unique_response = uamethod(unique_response)
         response = decorated_unique_response(ua.NodeId("ServerMethod", 2))
@@ -732,11 +734,11 @@ def unique_response(parent):
         self.assertEqual(type(response[0]), ua.Variant)
 
     def test_ua_method_complex_multiple(self):
-        register_extension_object(KeyValuePair)
+        register_extension_object(MyCustomClass)
 
         def unique_response(parent):
-            return [KeyValuePair("Key", "Value"),
-                    KeyValuePair("Hello", "World")]
+            return [MyCustomClass("Key", "Value"),
+                    MyCustomClass("Hello", "World")]
 
         decorated_unique_response = uamethod(unique_response)
         response = decorated_unique_response(ua.NodeId("ServerMethod", 2))
@@ -745,24 +747,24 @@ def unique_response(parent):
         self.assertEqual(type(response[0]), ua.Variant)
 
     # TODO: Fix the test
-    # def test_decode_custom_object(self):
-    #     register_extension_object(KeyValuePair)
-    #
-    #     def unique_response(parent):
-    #         return KeyValuePair("Key", "Value")
-    #
-    #     decorated_unique_response = uamethod(unique_response)
-    #     response = decorated_unique_response(ua.NodeId("ServerMethod", 2))
-    #
-    #     self.assertEqual(len(response), 1)
-    #     self.assertEqual(type(response[0]), ua.Variant)
-    #
-    #     key_value = uabin.unpack_uatype_array(response[0].VariantType, io.BytesIO(response[0].to_binary()))
-    #     self.assertEqual(key_value.key, "Key")
-    #     self.assertEqual(key_value.value, "Value")
-
-
-class KeyValuePair(object):
+    def test_decode_custom_object(self):
+        register_extension_object(MyCustomClass)
+
+        def unique_response(parent):
+            return MyCustomClass("Key", "Value")
+
+        decorated_unique_response = uamethod(unique_response)
+        response = decorated_unique_response(ua.NodeId("ServerMethod", 2))
+
+        self.assertEqual(len(response), 1)
+        self.assertEqual(type(response[0]), ua.Variant)
+
+        key_value = uabin.unpack_uatype_array(response[0].VariantType, io.BytesIO(response[0].to_binary()))
+        self.assertEqual(key_value.key, "Key")
+        self.assertEqual(key_value.value, "Value")
+
+
+class MyCustomClass(object):
     DEFAULT_BINARY = 20001
 
     def __init__(self, key, value, namespace_id=0):
@@ -782,4 +784,4 @@ def from_binary(data):
         namespace_index = uabin.Primitives.UInt16.unpack(data)
         key = uabin.Primitives.String.unpack(data)
         value = uabin.Primitives.String.unpack(data)
-        return KeyValuePair(key, value, namespace_index)
+        return MyCustomClass(key, value, namespace_index)