OPCUA: detect and cast custom method input-parameters

Problem

When we take a look at https://github.com/FreeOpcUa/opcua-asyncio/blob/master/asyncua/common/methods.py#L14, we see that the pattern to call an opcua method would be in the form of node.call_method('method_name', *args).

Now the protocol expect the args to be cast to the right datatypes. If you take a look at https://github.com/FreeOpcUa/opcua-asyncio/discussions/1278, you see that the custom datatype is pulled from ua and a value is built this way: ua._DB_Com_In_(ID=99, Value1=100, Value2=199).

But as interactive / dynamic client, I'm unable to know in advance what datatype I need... I would need something like:


args = cast_for_method('method_name', [{'ID': 99, 'Value1': 100, 'Value2': 199}, 10])
result = node.call_method('method_name'), *args)

What I tried

Get the type from the variable called 0:InputArguments of the method_name which is the place where the parameters will be stored.

I tried to call read_data_type(), read_data_type_as_variant_type() (this actually should return the right type but does not work for a method, I get all existing defined types instead).

read_data_value(), read_type_definition() and read_value() looks promising but I'm not there yet. I know that read_data_type_as_variant_type() is normally returning the datatype, therefore, I try to find an object with the above function that enable me to call this function so that I can finally get the type I need.

Solution

You need to read the value of the "0:InputArguments" with read_value. This returns a list of Arguments.

Each dataType in the Arguments can be translated to class like this:

from asyncua.ua import ObjectIds
from asyncua import ua
base_type_list = {
ua.NodeId(ObjectIds.Null): None,
    ua.NodeId(ObjectIds.Boolean): bool,
    ua.NodeId(ObjectIds.SByte): ua.SByte,
    ua.NodeId(ObjectIds.Byte): ua.Byte,
    ua.NodeId(ObjectIds.Int16): ua.Int16,
    ua.NodeId(ObjectIds.UInt16): ua.UInt16,
    ua.NodeId(ObjectIds.Int32): ua.Int32,
    ua.NodeId(ObjectIds.UInt32): ua.UInt32,
    ua.NodeId(ObjectIds.Int64): ua.Int64,
    ua.NodeId(ObjectIds.UInt64): ua.UInt64,
    ua.NodeId(ObjectIds.Float): ua.Float,
    ua.NodeId(ObjectIds.Double): ua.Double,
    ua.NodeId(ObjectIds.String): ua.String,
    ua.NodeId(ObjectIds.DateTime): ua.DateTime,
    ua.NodeId(ObjectIds.Guid): ua.Guid,
    ua.NodeId(ObjectIds.ByteString): ua.ByteString,
    ua.NodeId(ObjectIds.XmlElement): ua.XmlElement,
    ua.NodeId(ObjectIds.NodeId): ua.NodeId,
    ua.NodeId(ObjectIds.ExpandedNodeId): ua.ExpandedNodeId,
    ua.NodeId(ObjectIds.StatusCode): ua.StatusCode,
    ua.NodeId(ObjectIds.QualifiedName): ua.QualifiedName,
    ua.NodeId(ObjectIds.LocalizedText): ua.LocalizedText,
    ua.NodeId(ObjectIds.Structure): ua.ExtensionObject,
    ua.NodeId(ObjectIds.DataValue): ua.DataValue,
    ua.NodeId(ObjectIds.BaseVariableType): ua.Variant,
    ua.NodeId(ObjectIds.DiagnosticInfo): ua.DiagnosticInfo,
} 

def get_class_from_nodeid(datatype_nodeid):
  
   # check all type dicts   
   sources = [base_type_list, ua.basetype_by_datatype, ua.extension_objects_by_datatype, ua.enums_by_datatype]
   cls = None
   for src in sources:
       cls = src.get(datatype_nodeid, None)
       if cls is not None:
          return cls       
   return cls