pydra_test.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "import nest_asyncio\n",
    "\n",
    "nest_asyncio.apply()\n",
    "\n",
    "import pydra"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "@pydra.mark.task\n",
    "def add_var(a, b) ->{'sum_a_b': int} :\n",
    "    return a + b\n",
    "\n",
    "task1 = add_var()\n",
    "\n",
    "task1.inputs.a = 3\n",
    "task1.inputs.b = 6\n",
    "\n",
    "task1()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "## create workflows using pydra\n",
    "\n",
    "First we will define the functions used within the workflows\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@pydra.mark.task\n",
    "def add_var(a, b):\n",
    "    return a+b\n",
    "\n",
    "@pydra.mark.task\n",
    "def power(a, n=2):\n",
    "    return a**n\n",
    "\n",
    "@pydra.mark.task\n",
    "def mult_var(a,b):\n",
    "    return a*b\n",
    "\n",
    "@pydra.mark.task\n",
    "def add_two(a):\n",
    "    return a+2\n"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, we define some workflows to get a feeling of how to connect the different task within a workflow"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wf1 = pydra.Workflow(name= 'wf1', input_spec=['x','y'], x=3,y=4)\n",
    "\n",
    "wf1.add(add_var(name='sum_x_y', a=wf1.lzin.x, b=wf1.lzin.y))\n",
    "\n",
    "wf1.set_output([('out', wf1.sum_x_y.lzout.out)])\n",
    "\n",
    "with pydra.Submitter(plugin='cf') as sub:\n",
    "    sub(wf1)\n",
    "\n",
    "\n",
    "wf1.result()\n",
    "\n"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, that we have created a simple worflow lets do something slightly more complex and interconnect different task within a workflow(output of one task should be the input of another task)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wf2=pydra.Workflow(name= 'wf2', input_spec=['x','y'], x=3,y=4)  \n",
    "# Adding the two numbers\n",
    "wf2.add(add_var(name='sum_x_y', a=wf2.lzin.x, b=wf2.lzin.y))\n",
    "# Multiply the two numbers\n",
    "wf2.add(mult_var(name='mult_x_y', a=wf2.lzin.x, b=wf2.lzin.y))\n",
    "# take the values of the above operations and feed both of them into the power function using a splitter\n",
    "#wf2.add(power(name='power', a=[wf2.sum_x_y.lzout.sum_a_b, wf2.mult_x_y.lzout.out]).split('a'))\n",
    "wf2.add(power(name='power', a= wf2.sum_x_y.lzout.out))\n",
    "# adding two to the given output\n",
    "wf2.add(add_two(name='add_two', a=wf2.mult_x_y.lzout.out))\n",
    "\n",
    "wf2.set_output([('out_p', wf2.power.lzout.out),('out_other', wf2.add_two.lzout.out)])\n",
    "\n",
    "\n",
    "with pydra.Submitter(plugin='cf') as sub:\n",
    "    sub(wf2)\n",
    "\n",
    "\n",
    "wf2.result()\n",
    "\n"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### create workflows containing shell commands"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cmd = 'pwd'\n",
    "shellCommand = pydra.ShellCommandTask(name= 'printwd', executable=cmd)\n",
    "\n",
    "with pydra.Submitter(plugin='cf') as sub:\n",
    "    sub(shellCommand)\n",
    "\n",
    "\n",
    "shellCommand.result()\n",
    "# produces weird output for working directory, Check and issue on Github!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cmd =['echo','hail','pydra'] \n",
    "shellCommand = pydra.ShellCommandTask(name= 'printwd', executable=cmd)\n",
    "print('cmdline =', shellCommand.cmdline)\n",
    "\n",
    "with pydra.Submitter(plugin='cf') as sub:\n",
    "    sub(shellCommand)\n",
    "\n",
    "\n",
    "shellCommand.result()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create more specific (to the domain of neuroimaging) tasks and/or Workflows"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Opening raw data file /data2/egapontseva/MEG_QC_stuff/data/from openneuro/ds003483/sub-009/ses-1/meg/sub-009_ses-1_task-deduction_run-1_meg.fif...\n",
      "    Range : 60000 ... 1255999 =     60.000 ...  1255.999 secs\n",
      "Ready.\n",
      "Reading 0 ... 1195999  =      0.000 ...  1195.999 secs...\n"
     ]
    }
   ],
   "source": [
    "\n",
    "data_path = '/data2/egapontseva/MEG_QC_stuff/data/from openneuro/ds003483/sub-009/ses-1/meg/sub-009_ses-1_task-deduction_run-1_meg.fif'\n",
    "\n",
    "raw = mne.io.read_raw_fif(data_path, preload=True)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Opening raw data file /data2/egapontseva/MEG_QC_stuff/data/from openneuro/ds003483/sub-009/ses-1/meg/sub-009_ses-1_task-deduction_run-1_meg.fif...\n",
      "    Range : 60000 ... 1255999 =     60.000 ...  1255.999 secs\n",
      "Ready.\n",
      "Reading 0 ... 1195999  =      0.000 ...  1195.999 secs...\n",
      "Filtering raw data in 1 contiguous segment\n",
      "Setting up band-pass filter from 0.5 - 1e+02 Hz\n",
      "\n",
      "FIR filter parameters\n",
      "---------------------\n",
      "Designing a one-pass, zero-phase, non-causal bandpass filter:\n",
      "- Windowed time-domain design (firwin) method\n",
      "- Hamming window with 0.0194 passband ripple and 53 dB stopband attenuation\n",
      "- Lower passband edge: 0.50\n",
      "- Lower transition bandwidth: 0.50 Hz (-6 dB cutoff frequency: 0.25 Hz)\n",
      "- Upper passband edge: 100.00 Hz\n",
      "- Upper transition bandwidth: 25.00 Hz (-6 dB cutoff frequency: 112.50 Hz)\n",
      "- Filter length: 6601 samples (6.601 sec)\n",
      "\n",
      "Result(output=Output(filtered_data=<Raw | sub-009_ses-1_task-deduction_run-1_meg.fif, 320 x 1196000 (1196.0 s), ~2.86 GB, data loaded>), runtime=None, errored=False)\n"
     ]
    }
   ],
   "source": [
    "import pydra\n",
    "import mne\n",
    "from pydra.engine.task import TaskBase\n",
    "\n",
    "l_freq = 0.5\n",
    "h_freq = 100\n",
    "#data_path = '/data2/egapontseva/MEG_QC_stuff/data/from openneuro/ds000117/sub-01/ses-meg/meg/*run-01_meg.fif'\n",
    "\n",
    "# Define the Pydra workflow\n",
    "workflow_name = \"mne_filter_workflow\"\n",
    "input_spec = [\"data_path\", \"l_freq\", \"h_freq\"]\n",
    "mne_filter_wf = pydra.Workflow(name=workflow_name, input_spec=input_spec)\n",
    "\n",
    "# Load MEG data from file\n",
    "@pydra.mark.task\n",
    "def load_meg_data(data_path):\n",
    "    raw = mne.io.read_raw_fif(data_path, preload=True)\n",
    "    return raw\n",
    "@pydra.mark.task\n",
    "def apply_mne_filters(raw, l_freq, h_freq):\n",
    "    filt_raw = raw.copy().filter(l_freq, h_freq, fir_design='firwin')\n",
    "    return filt_raw\n",
    "\n",
    "\n",
    "# Load data with MNE\n",
    "mne_filter_wf.add(load_meg_data(name='load_meg_data', data_path= data_path))\n",
    "\n",
    "# Apply MNE filtering methods to the data\n",
    "\n",
    "mne_filter_wf.add(apply_mne_filters(name= 'filtering' , raw= mne_filter_wf.load_meg_data.lzout.out, l_freq =l_freq, h_freq=h_freq))\n",
    "\n",
    "# Specify the output of the workflow\n",
    "mne_filter_wf.set_output([\n",
    "    (\"filtered_data\", mne_filter_wf.filtering.lzout.out)\n",
    "])\n",
    "\n",
    "# Submit the workflow for execution using the 'cf' plugin\n",
    "with pydra.Submitter(plugin='cf') as sub:\n",
    "    sub(mne_filter_wf)\n",
    "\n",
    "# View the output of the workflow\n",
    "print(mne_filter_wf.result())\n"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The filtering workflow runs through lets now try some workflows with embedded megqc functions\n",
    "\n",
    "### MEGQC Workflows"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Using matplotlib as 2D backend.\n"
     ]
    }
   ],
   "source": [
    "import meg_qc  \n",
    "from meg_qc.meg_qc_pipeline import make_derivative_meg_qc\n",
    "\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "megqc",
   "language": "python",
   "name": "megqc"
  },
  "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.8.10"
  },
  "orig_nbformat": 4
 },
 "nbformat": 4,
 "nbformat_minor": 2
}