Retrieve the signature of prepare_shell_job_inputs to determine exp…

…ected input parameters.

aiida_workgraph/engine/utils.py

@@ -129,28 +129,16 @@ def prepare_for_python_task(task: dict, kwargs: dict, var_kwargs: dict) -> dict:
  return inputs
 
 
-def prepare_for_shell_task(task: dict, kwargs: dict) -> dict:
+def prepare_for_shell_task(task: dict, inputs: dict) -> dict:
  """Prepare the inputs for ShellJob"""
- from aiida_shell.launch import (
- convert_nodes_single_file_data,
- prepare_shell_job_inputs,
- )
+ from aiida_shell.launch import prepare_shell_job_inputs
+ import inspect
 
- metadata = kwargs.pop("metadata", {})
- monitors = kwargs.pop("monitors", {})
- nodes = convert_nodes_single_file_data(kwargs.pop("nodes", {}))
-
- metadata.update({"call_link_label": task["name"]})
-
- # find all keys in kwargs start with "nodes."
- for key in list(kwargs.keys()):
- if key.startswith("nodes."):
- nodes[key[6:]] = kwargs.pop(key)
-
- kwargs.update({"nodes": nodes, "metadata": metadata})
-
- inputs = prepare_shell_job_inputs(**kwargs)
-
- inputs["monitors"] = monitors
+ # Retrieve the signature of `prepare_shell_job_inputs` to determine expected input parameters.
+ signature = inspect.signature(prepare_shell_job_inputs)
+ kwargs = {key: inputs.pop(key, None) for key in signature.parameters.keys()}
+ inputs.update(prepare_shell_job_inputs(**kwargs))
 
+ inputs.setdefault("metadata", {})
+ inputs["metadata"].update({"call_link_label": task["name"]})
  return inputs