fix: polish headless

plantseg/tasks/io_tasks.py

@@ -13,10 +13,6 @@
             required=True,
             is_input_file=True,
         ),
-        "image_name": RunTimeInputSchema(
-            description="Name of the image (if None, the file name will be used)",
-            required=False,
-        ),
     },
 )
 def import_image_task(
@@ -68,7 +64,10 @@ def import_image_task(
         "export_directory": RunTimeInputSchema(
             description="Output directory path where the image will be saved", required=True
         ),
-        "name_pattern": RunTimeInputSchema(description="Output file name", required=False),
+        "name_pattern": RunTimeInputSchema(
+            description="Output file name pattern. Can contain the special {image_name} or {file_name} tokens ",
+            required=False,
+        ),
     },
 )
 def export_image_task(

plantseg/tasks/workflow_handler.py

@@ -78,15 +78,15 @@ class Task(BaseModel):
 
 class DAG(BaseModel):
     infos: Infos = Field(default_factory=Infos)
-    inputs: dict[str, Any] | list[dict[str, Any]] = Field(default_factory=dict)
+    inputs: list[dict[str, Any]] = Field(default_factory=lambda: [{}])
     list_tasks: list[Task] = Field(default_factory=list)
 
     """
     This model represents the Directed Acyclic Graph (DAG) of the workflow.
 
     Attributes:
         infos (Infos): A dictionary with the information of the workflow.
-        inputs (dict[str, Any]): A dictionary of the inputs of the workflow. For example path to the images and other runtime parameters.
+        inputs (list[dict[str, Any]): A dictionary of the inputs of the workflow. For example path to the images and other runtime parameters.
         list_tasks (list[Task]): A list of the tasks in the workflow.
 
     """
@@ -136,9 +136,9 @@ def prune_dag(dag: DAG) -> DAG:
         if task.id in reachable:
             new_dag.list_tasks.append(task)
 
-    for input_key, text in dag.inputs.items():
+    for input_key, text in dag.inputs[0].items():
         if input_key in reachable_inputs:
-            new_dag.inputs[input_key] = text
+            new_dag.inputs[0][input_key] = text
             new_dag.infos.inputs_schema[input_key] = dag.infos.inputs_schema[input_key]
 
     return new_dag
@@ -242,7 +242,7 @@ def _unique_input(name, id: int = 0):
         value_schema.task = func_name
         self._dag.infos.inputs_schema[unique_name] = value_schema
 
-        self._dag.inputs[unique_name] = value
+        self._dag.inputs[0][unique_name] = value
         return unique_name
 
     def clean_dag(self):

tests/headless/test_headless.py

@@ -64,6 +64,8 @@ def test_create_workflow(tmp_path):
         },
     ]
 
+    config['inputs'] = job_list
+
     with open(tmp_path / 'workflow.yaml', 'w') as file:
         yaml.dump(config, file)