presentation.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "216ba488-f950-4947-aee3-ffd9c39008ab",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "# Designing complex, reusable and scalable scientific workflows with Pydra\n",
    "\n",
    "> Ghislain VAILLANT, Inria"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "53638a6f",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [],
   "source": [
    "import nest_asyncio\n",
    "\n",
    "# Allow nested event loops.\n",
    "nest_asyncio.apply()\n",
    "\n",
    "# Reduce error backtrace size.\n",
    "%xmode Minimal"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ae90311d",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Motivation\n",
    "\n",
    "> Scientific workflows often require sophisticated analyses that encompass a large **collection of algorithms**. These algorithms are not necessarily designed to **work together** and are written by **different authors**.\n",
    ">\n",
    "> Some may be written in Python, while others might require calling **external programs**. It is common practice to create semi-manual workflows that require the scientists to **handle the files** and **interact with partial results** from algorithms and external tools.\n",
    ">\n",
    "> This approach is conceptually simple and easy to implement, but the resulting workflow is often **time consuming**, **error-prone** and **difficult to share** with others.\n",
    "> \n",
    "> -- <cite>[Pydra's Documentation](https://nipype.github.io/pydra/)</cite>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bd56fcc3",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## Challenges\n",
    "\n",
    "- Heterogeneity\n",
    "\n",
    "    _How to sequence computation from multiple tools using different data formats._\n",
    "\n",
    "- Reproducibility\n",
    "\n",
    "    _How difficult is it to share or re-use research code._\n",
    "\n",
    "- Scalability\n",
    "\n",
    "    _How much does code need to change to scale to larger computing power and storage._"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1a1ed594",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## Roadmap\n",
    "\n",
    "- Prerequisites\n",
    "- Core components\n",
    "- Advanced features\n",
    "- Support channels"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "603e2d06",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Prerequisites"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "569cc764",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "- Python 3.8+\n",
    "- Type annotations\n",
    "- Data classes"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fa2bd1b2",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Installation\n",
    "\n",
    "To install the core package:\n",
    "\n",
    "```shell\n",
    "$ pip install pydra==0.22\n",
    "```\n",
    "\n",
    "To install Pydra task packages, for instance ANTs:\n",
    "\n",
    "```shell\n",
    "$ pip install pydra-ants\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5a502db7",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Type annotations\n",
    "\n",
    "- Proposed in [PEP 484](https://peps.python.org/pep-0484/)\n",
    "- Available since Python 3.5\n",
    "- Implemented in syntax and [typing](https://docs.python.org/3/library/typing.html) module\n",
    "- Enhanced by subsequent Python releases"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "77cc2b6c",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Standard function definition."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e0757213",
   "metadata": {},
   "outputs": [],
   "source": [
    "def scale(factor, vector):\n",
    "    return [factor * x for x in vector]"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7f46b9b4",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "Definition with type annotations."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ceedb99d",
   "metadata": {},
   "outputs": [],
   "source": [
    "from typing import List\n",
    "\n",
    "# Type alias for convenience.\n",
    "Vector = List[float]\n",
    "\n",
    "def scale(factor: float, vector: Vector) -> Vector:\n",
    "    return [factor * x for x in vector]"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "504a9f84",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Data classes\n",
    "\n",
    "- Proposed in [PEP 557](https://peps.python.org/pep-0557/)\n",
    "- Available since Python 3.7\n",
    "- Implemented in [dataclasses](https://docs.python.org/3/library/dataclasses.html) module\n",
    "- Enhanced by third-party libraries such as [attrs](https://www.attrs.org/)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8e3e06ff",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Simple record definition."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bd9fb377-c668-4bac-9304-c7b1a0b1a9a8",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "fragment"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "import attrs\n",
    "\n",
    "@attrs.define\n",
    "class GeoPoint:\n",
    "    lat: float\n",
    "    lon: float"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "aa1487c2",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "swansea = GeoPoint(51.62, -3.94)\n",
    "\n",
    "print(swansea)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "471af07e",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Record with custom fields."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d8fe67c0",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "from attrs import define, field, validators\n",
    "\n",
    "def validate_lat(instance, attribute, value):\n",
    "    if abs(value) > 90:\n",
    "        raise ValueError(\n",
    "            f\"Latitude must be in range (-90, 90), got {value}.\")\n",
    "\n",
    "def validate_lon(instance, attribute, value):\n",
    "    if abs(value) > 180:\n",
    "        raise ValueError(\n",
    "            f\"Longitude must be in range (-180, 180), got {value}.\")\n",
    "\n",
    "@define(kw_only=True)   # Forbid init with posargs.\n",
    "class CustomGeoPoint:\n",
    "    lat: float = field(\n",
    "        validator=[validators.instance_of(float), validate_lat])\n",
    "\n",
    "    lon: float = field(\n",
    "        validator=[validators.instance_of(float), validate_lon])\n",
    "\n",
    "    alt: float = field(\n",
    "        default=0.0, metadata={\"recorded_by\": \"$DEVICE\"})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0522e312",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "swansea = CustomGeoPoint(lat=51.62, lon=-3.94)  # Okay!\n",
    "\n",
    "print(swansea)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "18667727",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "swansea = CustomGeoPoint(151.62, -3.94)             # Oops!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3d196c54",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "swansea = CustomGeoPoint(lat=151.62, lon=-3.94)     # Oops!"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6e6c2be0",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Core components\n",
    "\n",
    "Tasks, workflows and shell specifications."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "113ee694",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Python tasks"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "77544afa",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Defining a function task."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cbb962de",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pathlib import Path\n",
    "from pydra.mark import task\n",
    "\n",
    "# Define a Python task.\n",
    "@task\n",
    "def cwd() -> Path:\n",
    "    return Path.cwd()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8b1dfcf4",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Running a task."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7eb6e6f6",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Instantiate a task.\n",
    "task = cwd()\n",
    "\n",
    "# Run and get the results.\n",
    "result = task()\n",
    "\n",
    "print(result.output.out)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "251b109b",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Shell tasks"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6a8a6e69",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Defining a shell command task."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6608d43b",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pydra.engine.task import ShellCommandTask\n",
    "\n",
    "# Define a shell task.\n",
    "class Pwd(ShellCommandTask):\n",
    "    executable = \"pwd\"\n",
    "\n",
    "# Instantiate a task.\n",
    "task = Pwd()\n",
    "\n",
    "# Run and get the results.\n",
    "result = task()\n",
    "\n",
    "print(result.output.stdout)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f5a7a66e",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Defining input specifications."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7072f223",
   "metadata": {},
   "outputs": [],
   "source": [
    "from attrs import define, field\n",
    "from pydra.engine.specs import ShellSpec, SpecInfo\n",
    "\n",
    "# Define an input specifications.\n",
    "@define(kw_only=True)\n",
    "class InputSpec(ShellSpec):\n",
    "    level: int = field(\n",
    "        metadata={\"help_string\": \"max level\", \"argstr\": \"-L\"}\n",
    "    )\n",
    "    path: str = field(\n",
    "        metadata={\n",
    "            \"help_string\": \"input path\",\n",
    "            \"mandatory\": True,\n",
    "            \"argstr\": \"\",\n",
    "        }\n",
    "    )\n",
    "\n",
    "# Define the shell task.\n",
    "class Tree(ShellCommandTask):\n",
    "    executable = \"tree\"\n",
    "\n",
    "    # Associate the specifications with the task definition.\n",
    "    input_spec = SpecInfo(name=\"Inputs\", bases=(InputSpec,))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2d7c389e",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Testing input specifications."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a69f42af",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "from pathlib import Path\n",
    "\n",
    "# Instantiate a task.\n",
    "task = Tree(path=Path.cwd(), level=1)\n",
    "\n",
    "# Check the shell command.\n",
    "print(task.cmdline)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9b779b3b",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Output specifications."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "56ab970e",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "@define(kw_only=True)\n",
    "class InputSpec(ShellSpec):\n",
    "    input_file: Path = field(\n",
    "        metadata={\n",
    "            \"help_string\": \"input file\",\n",
    "            \"mandatory\": True,\n",
    "            \"argstr\": \"\",\n",
    "        }\n",
    "    )   \n",
    "    backup_file: str = field(\n",
    "        metadata={\n",
    "            \"help_string\": \"backup file\",\n",
    "            \"argstr\": \"\",\n",
    "            # Define the default file template.\n",
    "            \"output_file_template\": \"{input_file}.bak\",\n",
    "            # Whether to keep or strip file extensions.\n",
    "            \"keep_extension\": True,\n",
    "        }\n",
    "    )\n",
    "\n",
    "class Backup(ShellCommandTask):\n",
    "    executable = \"cp\"\n",
    "    input_spec = SpecInfo(name=\"Inputs\", bases=(InputSpec,))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3cda003b",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Testing output specifications."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "40424d73",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "# Instantiate a task.\n",
    "task = Backup(input_file=Path(\"./README.md\"))\n",
    "\n",
    "# Check the shell command.\n",
    "print(task.cmdline)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7719359e",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Workflows"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1b3ccb81",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Composing tasks in a workflow."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ee519f6e",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "from pydra.engine import Workflow\n",
    "\n",
    "workflow = Workflow(\n",
    "    name=\"my_workflow\", input_spec=[\"some_input\", \"other_input\"])\n",
    "\n",
    "# Add task and connect its inputs to the workflow inputs.\n",
    "workflow.add(\n",
    "    FooTask(name=\"foo_task\", foo=workflow.lzin.some_input))\n",
    "\n",
    "# Add task and connect its inputs to the workflow inputs.\n",
    "workflow.add(\n",
    "    BarTask(name=\"bar_task\", bar=workflow.lzin.other_input))\n",
    "\n",
    "# Add task and connect its inputs to upstream task outputs.\n",
    "workflow.add(\n",
    "    BazTask(name=\"baz_task\", baz=workflow.foo_task.lzout.out,\n",
    "            fred=workflow.bar_task.lzout.out))\n",
    "\n",
    "# Expose the workflow outputs.\n",
    "workflow.set_output({\"out\": workflow.baz_task.lzout.out})\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "610d6b2c",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Submitting a workflow for execution."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "50315272",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "from pydra.engine import Submitter\n",
    "\n",
    "# Setup an execution environment explicitly.\n",
    "with Submitter() as submitter:\n",
    "    # Submit the workflow for execution.\n",
    "    submitter(workflow)\n",
    "\n",
    "# Collect the results.\n",
    "result = workflow.result()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "09e9b7c5",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Summary\n",
    "\n",
    "- Define Python tasks with the `@task` and `@annotate`\n",
    "- Define shell tasks by subclassing `ShellCommandTask`\n",
    "- Parameterize shell tasks with `ShellSpec` and `ShellOutSpec`\n",
    "- Compose tasks into a workflow with `Workflow`"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "214eb688",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Advanced features"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5a592fe2",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Complex shell tasks"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ba96f1d9",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "Example with the `shasum` command.\n",
    "\n",
    "```shell\n",
    "$ shasum -h\n",
    "Usage: shasum [OPTION]... [FILE]...\n",
    "Print or check SHA checksums.\n",
    "With no FILE, or when FILE is -, read standard input.\n",
    "\n",
    "  -a, --algorithm   1 (default), 224, 256, 384, 512, 512224, 512256\n",
    "  -b, --binary      read in binary mode\n",
    "  -c, --check       read SHA sums from the FILEs and check them\n",
    "      --tag         create a BSD-style checksum\n",
    "  -t, --text        read in text mode (default)\n",
    "  -U, --UNIVERSAL   read in Universal Newlines mode\n",
    "                        produces same digest on Windows/Unix/Mac\n",
    "  -0, --01          read in BITS mode\n",
    "                        ASCII '0' interpreted as 0-bit,\n",
    "                        ASCII '1' interpreted as 1-bit,\n",
    "                        all other characters ignored\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c85666d4",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Mutually exclusive parameters with `xor`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "217aa36f",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "@define(kw_only=True)\n",
    "class SHASumSpec(ShellSpec):\n",
    "    input_file: Path = field(\n",
    "        metadata={\n",
    "            \"help_string\": \"input file\",\n",
    "            \"mandatory\": True,\n",
    "            \"argstr\": \"\",\n",
    "            \"position\": -1\n",
    "        }\n",
    "    )\n",
    "    ascii: bool = field(\n",
    "        metadata={\n",
    "            \"help_string\": \"read as ASCII bits\",\n",
    "            \"argstr\": \"-0\",\n",
    "            \"xor\": {\"binary\"},\n",
    "        }\n",
    "    ) \n",
    "    binary: bool = field(\n",
    "        metadata={\n",
    "            \"help_string\": \"read as binary\",\n",
    "            \"argstr\": \"-b\",\n",
    "            \"xor\": {\"ascii\"},\n",
    "        }\n",
    "    )\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "689a9f2e",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Choice parameters with `allowed_values`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0267afd3",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "@define(kw_only=True)\n",
    "class SHASumSpec(ShellSpec):\n",
    "    input_file: Path = field(\n",
    "        metadata={\n",
    "            \"help_string\": \"input file\",\n",
    "            \"mandatory\": True,\n",
    "            \"argstr\": \"\",\n",
    "            \"position\": -1\n",
    "        }\n",
    "    )\n",
    "    algorithm: int = field(\n",
    "        default=1,\n",
    "        metadata={\n",
    "            \"help_string\": \"algorithm (default is SHA-1)\",\n",
    "            \"argstr\": \"-a\",\n",
    "            # Restrict the domain of valid values.\n",
    "            \"allowed_values\": {1, 224, 256, 384, 512, 51224, 512256},\n",
    "        },\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "61d9fd92",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Custom formatting with `formatter`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1f5f857f",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "# Custom formatting logic for mode parameter.\n",
    "def format_mode(mode: str) -> str:\n",
    "    return {\"ascii\": \"-0\", \"binary\": \"-b\", \"text\": \"-t\"}.get(mode, \"\")\n",
    "\n",
    "@define(kw_only=True)\n",
    "class SHASumSpec(ShellSpec):\n",
    "    input_file: Path = field(\n",
    "        metadata={\n",
    "            \"help_string\": \"input file\",\n",
    "            \"mandatory\": True,\n",
    "            \"argstr\": \"\",\n",
    "            \"position\": -1\n",
    "        }\n",
    "    )\n",
    "    mode: str = field(\n",
    "        default=\"text\",\n",
    "        metadata={\n",
    "            \"help_string\": \"read mode\",\n",
    "            \"allowed_values\": [\"ascii\", \"binary\", \"text\"],\n",
    "            # argstr replaced by custom formatter.\n",
    "            \"formatter\": format_mode,\n",
    "        },\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0a885829",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Dependent parameters with `requires`.\n",
    "\n",
    "```sh\n",
    "$ shasum -h\n",
    "Usage: shasum [OPTION]... [FILE]...\n",
    "Print or check SHA checksums.\n",
    "With no FILE, or when FILE is -, read standard input.\n",
    "  ...\n",
    "  -c, --check       read SHA sums from the FILEs and check them\n",
    "  ...\n",
    "The following five options are useful only when verifying checksums:\n",
    "  ...\n",
    "  -w, --warn        warn about improperly formatted checksum lines\n",
    "  ...\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "dcacb0ba",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "@define(kw_only=True)\n",
    "class SHASumSpec(ShellSpec):\n",
    "    input_file: Path = field(\n",
    "        metadata={\n",
    "            \"help_string\": \"input file\",\n",
    "            \"mandatory\": True,\n",
    "            \"argstr\": \"\",\n",
    "            \"position\": -1\n",
    "        }\n",
    "    )\n",
    "    check: bool = field(\n",
    "        metadata={\n",
    "            \"help_string\": \"check sums from input file\",\n",
    "            \"argstr\": \"--check\",\n",
    "        }\n",
    "    )\n",
    "    warn: bool = field(\n",
    "        metadata={\n",
    "            \"help_string\": \"warn about faulty checksums\",\n",
    "            \"argstr\": \"--warn\",\n",
    "            \"requires\": {\"check\"},\n",
    "        }\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "18134d5c",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Custom output values"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e9ae5a05",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "from attrs import define, field\n",
    "from pydra.engine.specs import ShellOutSpec\n",
    "\n",
    "def parse_checksum(stdout: str) -> str:\n",
    "    return stdout.split(' ')[0]\n",
    "\n",
    "@define(kw_only=True)\n",
    "class SHASumOutSpec(ShellOutSpec):\n",
    "    checksum: str = field(\n",
    "        metadata={\n",
    "            \"help_string\": \"checksum of input file\",\n",
    "            \"callable\": parse_checksum,\n",
    "        }\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "21005b70",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "from attrs import field, define\n",
    "from pathlib import Path\n",
    "from pydra.engine.specs import ShellSpec, SpecInfo\n",
    "from pydra.engine.task import ShellCommandTask\n",
    "\n",
    "def format_mode(mode: str) -> str:\n",
    "    return {\"ascii\": \"-0\", \"binary\": \"-b\", \"text\": \"-t\"}.get(mode, \"\")\n",
    "\n",
    "@define(kw_only=True)\n",
    "class SHASumSpec(ShellSpec):\n",
    "    input_file: Path = field(\n",
    "        metadata={\n",
    "            \"help_string\": \"input file\",\n",
    "            \"mandatory\": True,\n",
    "            \"argstr\": \"\",\n",
    "            \"position\": -1\n",
    "        }\n",
    "    )\n",
    "    algorithm: int = field(\n",
    "        default=1,\n",
    "        metadata={\n",
    "            \"help_string\": \"algorithm (default is SHA-1)\",\n",
    "            \"argstr\": \"-a\",\n",
    "            # Restrict the domain of valid values.\n",
    "            \"allowed_values\": {1, 224, 256, 384, 512, 51224, 512256},\n",
    "        },\n",
    "    )\n",
    "    mode: str = field(\n",
    "        default=\"text\",\n",
    "        metadata={\n",
    "            \"help_string\": \"read mode\",\n",
    "            \"allowed_values\": [\"ascii\", \"binary\", \"text\"],\n",
    "            # argstr replaced by custom formatter.\n",
    "            \"formatter\": format_mode,\n",
    "        },\n",
    "    )\n",
    "\n",
    "class SHASum(ShellCommandTask):\n",
    "    executable = \"shasum\"\n",
    "    input_spec = SpecInfo(name=\"Inputs\", bases=(SHASumSpec,))\n",
    "    output_spec = SpecInfo(name=\"Ouputs\", bases=(SHASumOutSpec,))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f2d088d3",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "# Compute the SHA-256 sum of the readme file.\n",
    "task = SHASum(\n",
    "    input_file=Path.cwd() / \"README.md\",\n",
    "    algorithm=256,\n",
    "    mode=\"text\",\n",
    ")\n",
    "\n",
    "# Run the task and collect results.\n",
    "result = task()\n",
    "\n",
    "print(f\"stdout: \\t{result.output.stdout}\")\n",
    "print(f\"checksum: \\t{result.output.checksum}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c70a3b41",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Container tasks\n",
    "\n",
    "- Using the `container_info` parameter\n",
    "- Restricted to shell tasks\n",
    "- Docker or Singularity "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9b519bd1",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "Running a shell task in a Docker container:\n",
    "\n",
    "```python\n",
    "task = MyTask(..., container_info=(\"docker\", \"busybox\"))\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b2f2b6ef",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "Running a shell task in a Singularity container:\n",
    "\n",
    "```python\n",
    "task = MyTask(..., container_info=(\"singularity\", \"alpine\"))\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a280fd10",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "<div class=\"alert alert-info\">\n",
    "<b>Experimental</b>: Proposal for refactoring to a generic `Environment` class.\n",
    "\n",
    "https://github.com/nipype/pydra/pull/516\n",
    "</div>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f3bc6655",
   "metadata": {},
   "source": []
  },
  {
   "cell_type": "markdown",
   "id": "a0f1ed06",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Accelerate execution"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "14618a75",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Multicore task parallelism"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d946e317",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "from pydra.engine import Submitter\n",
    "\n",
    "# Scale computation to 2 cores.\n",
    "with Submitter(plugin=\"cf\", n_procs=2) as submitter:\n",
    "    submitter(workflow)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a69c2699",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Other plugins (Dask, SLURM, ...)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c8de607c",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "# Scale computation using Dask workers.\n",
    "with Submitter(plugin=\"dask\") as submitter:\n",
    "    submitter(workflow)\n",
    "\n",
    "# Scale computation using SLURM workers.\n",
    "with Submitter(plugin=\"slurm\") as submitter:\n",
    "    submitter(workflow)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "449ecdac",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "Other plugins are in the works, such as for [PSI/J](https://psi-j-python.readthedocs.io/)."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "eff715da",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Caching expensive computation\n",
    "\n",
    "```python\n",
    "task = MyTask(..., cache_dir=\"/path/to/cache/dir\")\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "133f463b",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Arbitrary workflows\n",
    "\n",
    "- Decompose workflows into subgraphs with `add`\n",
    "- Map over a list of parameter values with `split`\n",
    "- Merge results from mapped tasks with `combine`"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dc012d8e",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "<div class=\"alert alert-info\">\n",
    "<b>Discussion</b>: Syntax simplification of `split`, `combine` and `add`.\n",
    "\n",
    "https://github.com/nipype/pydra/discussions/692\n",
    "</div>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "106f9deb",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Summary\n",
    "\n",
    "- Specify complex shell tasks using `xor`, `requires`, `formatter` and `callable`\n",
    "- Execute shell tasks in a container using `container_info`\n",
    "- Scale workflow execution with `Submitter` plugins\n",
    "- Re-use intermediate computation with `cache_dir`\n",
    "- Design arbitrary workflow by nesting, splitting and combining"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c1fd3bcb",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Support channels\n",
    "\n",
    "- Documentation: `https://nipype.github.io/pydra`\n",
    "- Issues: `https://github.com/nipype/pydra/issues`\n",
    "- Discussions: `https://github.com/nipype/pydra/discussions`\n",
    "- Live chat: `https://mattermost.brainhack.org/brainhack/channels/nipype`\n",
    "- Cohacking: `https://meet.jit.si/pydra`"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d1035858",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Tutorial\n",
    "\n",
    "![The Pydra tutorial homepage](./assets/pydra-tutorial.png)\n",
    "\n",
    "> https://nipype.github.io/pydra-tutorial"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "482b20be",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Q&A\n",
    "\n",
    "![The NeuroStars homepage](./assets/neurostars-homepage.png)\n",
    "\n",
    "> https://neurostars.org"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d70a9da2",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Pydra task packages\n",
    "\n",
    "![Pydra task packages on PyPI](./assets/pypi-packages.png)\n",
    "\n",
    "> https://pypi.org/search/?q=pydra"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7fec457c",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## The next steps\n",
    "\n",
    "- Get comfortable with the core concepts behind Pydra\n",
    "- Write and run a first workflow you are very familiar with\n",
    "- Start with a linear dataflow, then increase in complexity\n",
    "- Do not hesitate to reach out for help to the community\n",
    "  \n",
    "> https://github.com/aramis-lab/RSECon23-Pydra"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "56de78c2",
   "metadata": {},
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}