Pass Socket in to a dynamic namespace raise an error

[superstar54]

Here is an example from @khsrali ,

s1 = wg.add_task(
            TransferCalculation,
            metadata= {
            'computer': Computer.get(label='localhost'),
            'description': 'Transfer calculation: upload file and retrieve all contents',
            'options': {
                'resources': {'num_machines': 1}, 
            },},
            source_nodes= {
                'local': input_file,
            },

            instructions= Dict({
                'local_files': [['local', FILENAME, FILENAME]],
                'retrieve_files': False
            })
            )
s2 = wg.add_task(
            TransferCalculation,
            metadata= {
            'computer': Computer.get(label='localhost'),
            'description': 'Transfer calculation: upload file and retrieve all contents --2',
            'options': {
                'resources': {'num_machines': 1},
            },},
            source_nodes= {
                'remote': s1.outputs["remote_folder"],
            },

            instructions= Dict({
                'remote_files': [['remote', FILENAME, FILENAME]],
                'retrieve_files': True
            })
            )

and, it gives the following error:

    raise ConstructorError("while constructing a Python object", mark,
yaml.constructor.ConstructorError: while constructing a Python object
cannot find 'DecoratedNode' in the module 'node_graph.utils'
  in "<unicode string>", line 1243, column 19:
              parent: &id003 !!python/object:node_grap ... 
                      ^

The source_nodes input is a dynamic namespace, and in AiiDA, we can pass any data into it, e.g., the remote input as shown above, but there is not source_nodes.remote input in the task, so the WorkGraph serialize the whole dict as a data node.

Solution

Here is a temporary solution: we add a input for the task manually:

s2=wg.add_task(
            TransferCalculation,
            metadata= {
            'computer': Computer.get(label='localhost'),
            'description': 'Transfer calculation: upload file and retrieve all contents --2',
            'options': {
                'resources': {'num_machines': 1},
            },},
            instructions= Dict({
                'remote_files': [['remote', FILENAME, FILENAME]],
                'retrieve_files': True
            })
            )
        # add a `source_nodes.remote` input manually.
        s2.inputs.new('workgraph.any', name="source_nodes.remote")
        s2.kwargs.append("source_nodes.remote")
        s2.set({"source_nodes.remote": s1.outputs["remote_folder"]})