[ENH] add utility function to compute expected value

docs/pipelines.md

@@ -124,6 +124,16 @@ As explained before, all pipeline inherit from the `narps_open.pipelines.Pipelin
 * `fwhm` : full width at half maximum for the smoothing kernel (in mm) :
 * `tr` : repetition time of the fMRI acquisition (equals 1.0s)
 
+## Utility functions
+
+You can find several utility functions in the `narps_open.utils` module. 
+
+- raw_data_template
+- fmriprep_data_template
+- compute_expected_value
+
+Feel free to use them in your pipeline.
+
 ## Test your pipeline
 
 First have a look at the [testing topic of the documentation](./docs/testing.md). It explains how testing works for inside project and how you should write the tests related to your pipeline.

narps_open/utils/__init__.py

@@ -1,13 +1,17 @@
 #!/usr/bin/python
 # coding: utf-8
+from __future__ import annotations
 
 """ A set of utils functions for the narps_open package """
 
+from hashlib import sha256
 from os import listdir
-from os.path import isfile, join, abspath, dirname, realpath, splitext
+from os.path import abspath, dirname, isfile, join, realpath, splitext
+from pathlib import Path
 
+import pandas as pd
 from nibabel import load
-from hashlib import sha256
+
 
 def show_download_progress(count, block_size, total_size):
     """ A hook function to be passed to urllib.request.urlretrieve in order to
@@ -192,3 +196,48 @@ def fmriprep_data_template() -> dict:
     )
 
     return {"func_preproc": func_preproc, "confounds_file": confounds_file}
+
+
+def compute_expected_value(onsets: dict[str, list[float]] | pd.DataFrame | str | Path):
+    """Compute expected value regressor for a run.
+
+    Parameters
+    ----------
+    onsets : dict[str, list[float]] | pd.DataFrame | str | Path
+             Events for a run.
+             Pathlike TSV file with columns 'gain' and 'loss'.
+             If a dict, must have keys 'gain' and 'loss'.
+             If a DataFrame, must have columns 'gain' and 'loss'.
+
+    Returns
+    -------
+    onsets : pd.DataFrame
+             Onsets with expected value column added.
+    """
+    # # compute euclidian distance to the indifference line defined by
+    # # gain twice as big as losses
+    # % https://en.wikipedia.org/wiki/Distance_from_a_point_to_a_line
+    # a = 0.5;
+    # b = -1;
+    # c = 0;
+    # x = onsets{iRun#.gain;
+    # y = onsets{iRun}.loss;
+    # dist = abs(a * x + b * y + c) / (a^2 + b^2)^.5;
+    # onsets{iRun}.EV = dist; % create an "expected value" regressor
+    a = 0.5
+    b = -1
+    c = 0
+
+    if isinstance(onsets, (str, Path)):
+        onsets = pd.read_csv(onsets, sep="\t")
+
+    if isinstance(onsets, dict):
+        onsets = pd.DataFrame(onsets)
+
+    x = onsets["gain"]
+    y = onsets["loss"]
+
+    dist = abs(a * x + b * y + c) / (a**2 + b**2) ** 0.5
+    onsets["EV"] = dist
+
+    return onsets

narps_open/utils/utils.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pandas as pd
+
+# # compute euclidian distance to the indifference line defined by
+# # gain twice as big as losses
+# % https://en.wikipedia.org/wiki/Distance_from_a_point_to_a_line
+# a = 0.5;
+# b = -1;
+# c = 0;
+# x = onsets{iRun#.gain;
+# y = onsets{iRun}.loss;
+# dist = abs(a * x + b * y + c) / (a^2 + b^2)^.5;
+# onsets{iRun}.EV = dist; % create an "expected value" regressor
+
+
+def compute_expected_value(onsets: dict[str, list[float]] | pd.DataFrame | str | Path):
+    """Compute expected value regressor for a run.
+
+    Parameters
+    ----------
+    onsets : dict[str, list[float]] | pd.DataFrame | str | Path
+             Events for a run.
+             Pathlike TSV file with columns 'gain' and 'loss'.
+             If a dict, must have keys 'gain' and 'loss'.
+             If a DataFrame, must have columns 'gain' and 'loss'.
+
+    Returns
+    -------
+    onsets : pd.DataFrame
+             Onsets with expected value column added.
+    """
+    a = 0.5
+    b = -1
+    c = 0
+
+    if isinstance(onsets, (str, Path)):
+        onsets = pd.read_csv(onsets, sep="\t")
+
+    if isinstance(onsets, dict):
+        onsets = pd.DataFrame(onsets)
+
+    x = onsets["gain"]
+    y = onsets["loss"]
+
+    dist = abs(a * x + b * y + c) / (a**2 + b**2) ** 0.5
+    onsets["EV"] = dist
+
+    return onsets

tests/utils/test_utils.py

@@ -12,36 +12,62 @@
 """
 from os.path import join
 
+import pandas as pd
+from numpy.testing import assert_array_almost_equal
 from pytest import mark
 
+from narps_open.utils import (compute_expected_value, hash_dir_images,
+                              hash_image, show_download_progress)
 from narps_open.utils.configuration import Configuration
-from narps_open.utils import show_download_progress, hash_image, hash_dir_images
+
 
 class TestUtils:
-    """ A class that contains all the unit tests for the utils module."""
+    """A class that contains all the unit tests for the utils module."""
 
     @staticmethod
     @mark.unit_test
-    def test_show_download_progress(capfd): # using pytest's capfd fixture to get stdout
-        """ Test the show_download_progress function """
-
-        show_download_progress(25,1,100)
+    def test_show_download_progress(
+        capfd,
+    ):  # using pytest's capfd fixture to get stdout
+        """Test the show_download_progress function."""
+        show_download_progress(25, 1, 100)
         captured = capfd.readouterr()
-        assert captured.out == 'Downloading 25 %\r'
+        assert captured.out == "Downloading 25 %\r"
 
-        show_download_progress(26,2,200)
+        show_download_progress(26, 2, 200)
         captured = capfd.readouterr()
-        assert captured.out == 'Downloading 26 %\r'
+        assert captured.out == "Downloading 26 %\r"
 
-        show_download_progress(25,50,-1)
+        show_download_progress(25, 50, -1)
         captured = capfd.readouterr()
-        assert captured.out == 'Downloading ⣽\r'
+        assert captured.out == "Downloading ⣽\r"
 
     @staticmethod
     @mark.unit_test
-    def test_hash_image():
-        """ Test the hash_image function """
+    def test_compute_expected_value(tmp_path):
+        """Test the compute_expected_value function."""
+        onsets = {"gain": [1, 2, 3], "loss": [1, 2, 3]}
+
+        computed = compute_expected_value(onsets=onsets)
+        assert_array_almost_equal(computed["EV"], [0.447, 0.894, 1.342], decimal=3)
+
+        df = pd.DataFrame(onsets)
+        computed = compute_expected_value(onsets=df)
+        assert_array_almost_equal(computed["EV"], [0.447, 0.894, 1.342], decimal=3)
+
+        events_tsv = tmp_path / "events.tsv"
+        df.to_csv(events_tsv, sep="\t")
+        computed = compute_expected_value(onsets=events_tsv)
+        assert_array_almost_equal(computed["EV"], [0.447, 0.894, 1.342], decimal=3)
 
+        events_tsv = str(events_tsv)
+        computed = compute_expected_value(onsets=events_tsv)
+        assert_array_almost_equal(computed["EV"], [0.447, 0.894, 1.342], decimal=3)
+
+    @staticmethod
+    @mark.unit_test
+    def test_hash_image():
+        """Test the hash_image function."""
         # Get test_data for hash
         test_image_path = join(
             Configuration()['directories']['test_data'],
@@ -54,8 +80,7 @@ def test_hash_image():
     @staticmethod
     @mark.unit_test
     def test_hash_dir_images():
-        """ Test the hash_dir_images function """
-
+        """Test the hash_dir_images function."""
         # Get test_data for hash
         test_path = join(
             Configuration()['directories']['test_data'],