debugged and formatted tutorial notebooks

Author: tclose

new-docs/source/tutorial/3-troubleshooting.ipynb

@@ -36,6 +36,7 @@
    "source": [
     "# This is needed to run parallel workflows in Jupyter notebooks\n",
     "import nest_asyncio\n",
+    "\n",
     "nest_asyncio.apply()"
    ]
   },
@@ -126,14 +127,14 @@
    "outputs": [],
    "source": [
     "# This workflow will fail because we are trying to divide by 0\n",
-    "wf = UnsafeDivisionWorkflow(a=10, b=5).split(denominator=[3, 2 ,0])\n",
+    "wf = UnsafeDivisionWorkflow(a=10, b=5).split(denominator=[3, 2, 0])\n",
     "\n",
     "if __name__ == \"__main__\":\n",
     "    try:\n",
     "        with Submitter(worker=\"cf\") as sub:\n",
     "            result = sub(wf)\n",
     "    except Exception as e:\n",
-    "        print(e)\n"
+    "        print(e)"
    ]
   },
   {
@@ -156,7 +157,9 @@
     "from pprint import pprint\n",
     "from pydra.tasks.testing import Divide\n",
     "\n",
-    "with open(default_run_cache_dir / Divide(x=15, y=0)._checksum / \"_error.pklz\", \"rb\") as f:\n",
+    "with open(\n",
+    "    default_run_cache_dir / Divide(x=15, y=0)._checksum / \"_error.pklz\", \"rb\"\n",
+    ") as f:\n",
     "    error = cp.load(f)\n",
     "\n",
     "pprint(error)"
@@ -196,12 +199,12 @@
     "from pydra.engine.submitter import Submitter\n",
     "from pydra.tasks.testing import SafeDivisionWorkflow\n",
     "\n",
-    "wf = SafeDivisionWorkflow(a=10, b=5).split(denominator=[3, 2 ,0])\n",
+    "wf = SafeDivisionWorkflow(a=10, b=5).split(denominator=[3, 2, 0])\n",
     "\n",
     "if __name__ == \"__main__\":\n",
     "    with Submitter(worker=\"cf\") as sub:\n",
     "        result = sub(wf)\n",
-    "    \n",
+    "\n",
     "print(f\"Workflow completed successfully, results saved in: {result.output_dir}\")"
    ]
   },
@@ -229,9 +232,10 @@
     "        result = cp.load(f)\n",
     "    if result.outputs is not None:\n",
     "        for field_name in result.outputs:\n",
-    "            if result.outputs[field_name] == float('nan'):\n",
-    "                print(f\"Task {task_cache_dir.name!r} produced a NaN value for {field_name!r}\")\n",
-    "    "
+    "            if result.outputs[field_name] == float(\"nan\"):\n",
+    "                print(\n",
+    "                    f\"Task {task_cache_dir.name!r} produced a NaN value for {field_name!r}\"\n",
+    "                )"
    ]
   }
  ],

new-docs/source/tutorial/5-shell.ipynb

@@ -260,12 +260,14 @@
     "        \"--int-arg <int_arg:int?> \"\n",
     "        \"--tuple-arg <tuple_arg:int,str> \"\n",
     "    ),\n",
-    "    inputs={\"recursive\": shell.arg(\n",
-    "        help=(\n",
-    "            \"If source_file designates a directory, cp copies the directory and \"\n",
-    "            \"the entire subtree connected at that point.\"\n",
+    "    inputs={\n",
+    "        \"recursive\": shell.arg(\n",
+    "            help=(\n",
+    "                \"If source_file designates a directory, cp copies the directory and \"\n",
+    "                \"the entire subtree connected at that point.\"\n",
+    "            )\n",
     "        )\n",
-    "    )},\n",
+    "    },\n",
     "    outputs={\n",
     "        \"out_dir\": shell.outarg(position=-2),\n",
     "        \"out_file\": shell.outarg(position=-1),\n",
@@ -297,7 +299,8 @@
     "from pathlib import Path\n",
     "from fileformats.generic import File\n",
     "\n",
-    "# Arguments to the callable function can be one of \n",
+    "\n",
+    "# Arguments to the callable function can be one of\n",
     "def get_file_size(out_file: Path) -> int:\n",
     "    \"\"\"Calculate the file size\"\"\"\n",
     "    result = os.stat(out_file)\n",
@@ -356,6 +359,7 @@
     "from pydra.utils.typing import MultiInputObj\n",
     "from fileformats.generic import FsObject, Directory\n",
     "\n",
+    "\n",
     "@shell.define\n",
     "class Cp(ShellDef[\"Cp.Outputs\"]):\n",
     "\n",
@@ -369,7 +373,7 @@
     "\n",
     "    @shell.outputs\n",
     "    class Outputs(ShellOutputs):\n",
-    "        out_dir: Directory = shell.outarg(path_template=\"{out_dir}\")\n"
+    "        out_dir: Directory = shell.outarg(path_template=\"{out_dir}\")"
    ]
   },
   {
@@ -396,20 +400,18 @@
     "        \"in_file\": shell.arg(type=File, help=\"output file\", argstr=\"\", position=-2)\n",
     "    },\n",
     "    outputs={\n",
-    "        \"out_file\": shell.outarg(\n",
-    "            type=File, help=\"output file\", argstr=\"\", position=-1\n",
-    "        ),\n",
+    "        \"out_file\": shell.outarg(type=File, help=\"output file\", argstr=\"\", position=-1),\n",
     "        \"out_file_size\": {\n",
     "            \"type\": int,\n",
     "            \"help\": \"size of the output directory\",\n",
     "            \"callable\": get_file_size,\n",
-    "        }\n",
+    "        },\n",
     "    },\n",
     ")\n",
     "\n",
     "\n",
     "print(f\"ACommand input fields: {list_fields(ACommand)}\")\n",
-    "print(f\"ACommand input fields: {list_fields(ACommand.Outputs)}\")\n"
+    "print(f\"ACommand input fields: {list_fields(ACommand.Outputs)}\")"
    ]
   },
   {

new-docs/source/tutorial/6-workflow.ipynb

@@ -30,6 +30,7 @@
    "source": [
     "from pydra.design import workflow, python\n",
     "\n",
+    "\n",
     "# Example python task definitions\n",
     "@python.define\n",
     "def Add(a, b):\n",
@@ -83,6 +84,7 @@
     "from pydra.design import shell\n",
     "from fileformats import image, video\n",
     "\n",
+    "\n",
     "@workflow.define\n",
     "def ShellWorkflow(\n",
     "    input_video: video.Mp4,\n",
@@ -129,6 +131,7 @@
     "def Sum(x: list[float]) -> float:\n",
     "    return sum(x)\n",
     "\n",
+    "\n",
     "@workflow.define\n",
     "def SplitWorkflow(a: list[int], b: list[float]) -> list[float]:\n",
     "    # Multiply over all combinations of the elements of a and b, then combine the results\n",
@@ -232,6 +235,7 @@
     "def Subtract(x: float, y: float) -> float:\n",
     "    return x - y\n",
     "\n",
+    "\n",
     "@workflow.define\n",
     "def RecursiveNestedWorkflow(a: float, depth: int) -> float:\n",
     "    add = workflow.add(Add(x=a, y=1))\n",
@@ -292,6 +296,7 @@
     "    \"--width <width:int> --height <height:int>\",\n",
     ")\n",
     "\n",
+    "\n",
     "@workflow.define\n",
     "def TypeErrorWorkflow(\n",
     "    input_video: video.Mp4,\n",
@@ -313,7 +318,9 @@
     "\n",
     "    try:\n",
     "        handbrake = workflow.add(\n",
-    "            QuicktimeHandbrake(in_video=add_watermark.out_video, width=1280, height=720),\n",
+    "            QuicktimeHandbrake(\n",
+    "                in_video=add_watermark.out_video, width=1280, height=720\n",
+    "            ),\n",
     "        )  # This will raise a TypeError because the input video is an Mp4\n",
     "    except TypeError:\n",
     "        handbrake = workflow.add(\n",

new-docs/source/tutorial/7-canonical-form.ipynb

@@ -79,6 +79,7 @@
     "    def function(a, b):\n",
     "        return a + b, a / b\n",
     "\n",
+    "\n",
     "pprint(fields_dict(CanonicalPythonDef))\n",
     "pprint(fields_dict(CanonicalPythonDef.Outputs))"
    ]
@@ -127,6 +128,7 @@
     "    def function(a, b):\n",
     "        return a + b, a / b\n",
     "\n",
+    "\n",
     "pprint(fields_dict(CanonicalPythonDef))\n",
     "pprint(fields_dict(CanonicalPythonDef.Outputs))"
    ]
@@ -174,7 +176,7 @@
     "            result = os.stat(out_file)\n",
     "            return result.st_size\n",
     "\n",
-    "        out_file: generic.File\n",
+    "        copied: generic.FsObject = shell.outarg(path_template=\"copied\")\n",
     "        out_file_size: int = shell.out(callable=get_file_size)\n",
     "\n",
     "\n",
@@ -201,6 +203,7 @@
     "from pydra.design import python, workflow\n",
     "from pydra.engine.specs import WorkflowDef, WorkflowOutputs\n",
     "\n",
+    "\n",
     "# Example python task definitions\n",
     "@python.define\n",
     "def Add(a, b):\n",

pydra/engine/helpers.py

@@ -662,8 +662,12 @@ def fields_in_formatter(formatter: str | ty.Callable[..., str]) -> set[str]:
     """Extract all field names from a formatter string or function."""
     if isinstance(formatter, str):
         return parse_format_string(formatter)
-    else:
+    elif isinstance(formatter, ty.Sequence):
+        return set().union(*[fields_in_formatter(f) for f in formatter])
+    elif isinstance(formatter, ty.Callable):
         return set(inspect.signature(formatter).parameters.keys())
+    else:
+        raise ValueError(f"Unsupported formatter type: {type(formatter)} ({formatter})")
 
 
 def ensure_list(obj, tuple2list=False):

pydra/engine/specs.py

@@ -298,16 +298,13 @@ def __call__(
                 time_of_crash = result.errors["time of crash"]
                 error_message = "\n".join(result.errors["error message"])
             else:
-                time_of_crash = "UNKNOWN"
+                time_of_crash = "UNKNOWN-TIME"
                 error_message = "NOT RETRIEVED"
             raise RuntimeError(
                 f"Task {self} failed @ {time_of_crash} with the "
-                "following errors:\n"
-                + error_message
-                + (
-                    "To inspect, please load the pickled task object from here: "
-                    f"{result.output_dir}/_task.pklz"
-                )
+                f"following errors:\n{error_message}\n"
+                "To inspect, please load the pickled task object from here: "
+                f"{result.output_dir}/_task.pklz"
             )
         return result.outputs
 
@@ -835,9 +832,9 @@ def _from_task(cls, task: "Task[WorkflowDef]") -> Self:
                         time_of_crash = "UNKNOWN-TIME"
                         error_message = "NOT RETRIEVED"
                     errors.append(
-                        f"Task {node.name!r} failed @ {time_of_crash} "
-                        f"running {node._definition} with the following errors:\n{error_message}"
-                        "To inspect, please load the pickled task object from here: "
+                        f"Task {node.name!r} failed @ {time_of_crash} running "
+                        f"{node._definition} with the following errors:\n{error_message}"
+                        "\nTo inspect, please load the pickled task object from here: "
                         f"{result.output_dir}/_task.pklz"
                     )
             raise RuntimeError(