docs/changes/newsfragments/182.enh

@@ -0,0 +1 @@
+Support element(s) to be specified via text file for ``--element`` option of ``junifer run`` by `Synchon Mandal`_

docs/using/running.rst

@@ -29,20 +29,68 @@ The ``run`` command accepts the following additional arguments:
 * ``--element``: The *element* to run. If not specified, all elements will be
   run. This parameter can be specified multiple times to run multiple elements.
   If the *element* requires several parameters, they can be specified by
-  separating them with ``,``.
+  separating them with ``,``. It also accepts a file (e.g., ``elements.txt``)
+  containing complete or partial element(s).
 
 Example of running two elements:
+--------------------------------
 
 .. code-block:: bash
 
     junifer run config.yaml --element sub-01 --element sub-02
 
+You can also specify the elements via a text file like so:
+
+.. code-block:: bash
+
+    junifer run config.yaml --element elements.txt
+
+And the corresponding ``elements.txt`` would be like so:
+
+.. code-block:: text
+
+    sub-01
+    sub-02
+
 Example of elements with multiple parameters and verbose output:
+----------------------------------------------------------------
 
 .. code-block:: bash
 
     junifer run --verbose info config.yaml --element sub-01,ses-01
 
+You can also specify the elements via a text file like so:
+
+.. code-block:: bash
+
+    junifer run --verbose info config.yaml --element elements.txt
+
+And the corresponding ``elements.txt`` would be like so:
+
+.. code-block:: text
+
+    sub-01,ses-01
+
+In case you wanted to run for all possible sessions (e.g., ``ses-01``,
+``ses-02``, ``ses-03``) but only for ``sub-01``, you could also do:
+
+.. code-block:: bash
+
+    junifer run --verbose info config.yaml --element sub-01
+
+or,
+
+.. code-block:: bash
+
+    junifer run --verbose info config.yaml --element elements.txt
+
+and then the ``elements.txt`` would be like so:
+
+.. code-block:: text
+
+    sub-01
+
+
 .. _collect:
 
 Collecting Results

junifer/api/cli.py

@@ -8,9 +8,10 @@ import pathlib
 import subprocess
 import sys
 from pathlib import Path
-from typing import Dict, List, Union
+from typing import Dict, List, Tuple, Union
 
 import click
+import pandas as pd
 
 from ..utils.logging import (
     configure_logging,
@@ -32,27 +33,45 @@ from .utils import (
 )
 
 
-def _parse_elements(element: str, config: Dict) -> Union[List, None]:
+def _parse_elements(element: Tuple[str], config: Dict) -> Union[List, None]:
     """Parse elements from cli.
 
     Parameters
     ----------
-    element : str
-        The element to operate on.
+    element : tuple of str
+        The element(s) to operate on.
     config : dict
         The configuration to operate using.
 
     Returns
     -------
-    list
-        The element(s) as list.
+    list or None
+        The element(s) as list or None.
+
+    Raises
+    ------
+    ValueError
+        If no element is found either in the command-line options or
+        the configuration file.
+
+    Warns
+    -----
+    RuntimeWarning
+        If elements are specified both via the command-line options and
+        the configuration file.
 
     """
     logger.debug(f"Parsing elements: {element}")
+    # Early return None to continue with all elements
     if len(element) == 0:
         return None
-    # TODO: If len == 1, check if its a file, then parse elements from file
-    elements = [tuple(x.split(",")) if "," in x else x for x in element]
+    # Check if the element is a file for single element;
+    # if yes, then parse elements from it
+    if len(element) == 1 and Path(element[0]).resolve().is_file():
+        elements = _parse_elements_file(Path(element[0]).resolve())
+    else:
+        # Process multi-keyed elements
+        elements = [tuple(x.split(",")) if "," in x else x for x in element]
     logger.debug(f"Parsed elements: {elements}")
     if elements is not None and "elements" in config:
         warn_with_log(
@@ -61,19 +80,49 @@ def _parse_elements(element: str, config: Dict) -> Union[List, None]:
             "over the configuration file. That is, the elements specified "
             "in the command line will be used. The elements specified in "
             "the configuration file will be ignored. To remove this warning, "
-            'please remove the "elements" item from the configuration file.'
+            "please remove the `elements` item from the configuration file."
         )
     elif elements is None:
+        # Check in config
         elements = config.get("elements", None)
         if elements is None:
             raise_error(
-                "The 'elements' key is set in the configuration, but its value"
-                " is 'None'. It is likely that there is an empty 'elements' "
+                "The `elements` key is set in the configuration, but its value"
+                " is `None`. It is likely that there is an empty `elements` "
                 "section in the yaml configuration file."
             )
     return elements
 
 
+def _parse_elements_file(filepath: Path) -> List[Tuple[str, ...]]:
+    """Parse elements from file.
+
+    Parameters
+    ----------
+    filepath : pathlib.Path
+        The path to the element file.
+
+    Returns
+    -------
+    list of tuple of str
+        The element(s) as list.
+
+    """
+    # Read CSV into dataframe
+    csv_df = pd.read_csv(
+        filepath,
+        header=None,  # no header  # type: ignore
+        index_col=False,  # no index column
+        skipinitialspace=True,  # no leading space after delimiter
+    )
+    # Remove trailing whitespace in cell entries
+    csv_df_trimmed = csv_df.apply(
+        lambda x: x.str.strip() if x.dtype == "object" else x
+    )
+    # Convert to list of tuple of str
+    return list(map(tuple, csv_df_trimmed.to_numpy().astype(str)))
+
+
 def _validate_verbose(
     ctx: click.Context, param: str, value: str
 ) -> Union[str, int]:
@@ -133,7 +182,9 @@ def cli() -> None:  # pragma: no cover
     callback=_validate_verbose,
     default="info",
 )
-def run(filepath: click.Path, element: str, verbose: Union[str, int]) -> None:
+def run(
+    filepath: click.Path, element: Tuple[str], verbose: Union[str, int]
+) -> None:
     """Run command for CLI.
 
     \f
@@ -142,8 +193,8 @@ def run(filepath: click.Path, element: str, verbose: Union[str, int]) -> None:
     ----------
     filepath : click.Path
         The filepath to the configuration file.
-    element : str
-        The element to operate using.
+    element : tuple of str
+        The element to operate on.
     verbose : click.Choice
         The verbosity level: warning, info or debug (default "info").
 

junifer/api/functions.py

@@ -163,7 +163,9 @@ def run(
     # Fit elements
     with datagrabber_object:
         if elements is not None:
-            for t_element in elements:
+            for t_element in datagrabber_object.filter(
+                elements  # type: ignore
+            ):
                 mc.fit(datagrabber_object[t_element])
         else:
             for t_element in datagrabber_object:

junifer/api/tests/test_cli.py

@@ -5,13 +5,13 @@
 # License: AGPL
 
 from pathlib import Path
-from typing import Tuple
+from typing import List, Tuple
 
 import pytest
 from click.testing import CliRunner
 from ruamel.yaml import YAML
 
-from junifer.api.cli import collect, run, selftest, wtf
+from junifer.api.cli import _parse_elements_file, collect, run, selftest, wtf
 
 
 # Configure YAML class
@@ -35,7 +35,16 @@ runner = CliRunner()
 def test_run_and_collect_commands(
     tmp_path: Path, elements: Tuple[str, ...]
 ) -> None:
-    """Test run and collect commands."""
+    """Test run and collect commands.
+
+    Parameters
+    ----------
+    tmp_path : pathlib.Path
+        The path to the test directory.
+    elements : tuple of str
+        The parametrized elements to operate on.
+
+    """
     # Get test config
     infile = Path(__file__).parent / "data" / "gmd_mean.yaml"
     # Read test config
@@ -74,6 +83,104 @@ def test_run_and_collect_commands(
     assert collect_result.exit_code == 0
 
 
+@pytest.mark.parametrize(
+    "elements",
+    [
+        "sub-01",
+        "sub-01\nsub-02",
+        "    sub-01    ",
+        "sub-01\n    sub-02",
+    ],
+)
+def test_run_using_element_file(tmp_path: Path, elements: str) -> None:
+    """Test run command using element file.
+
+    Parameters
+    ----------
+    tmp_path : pathlib.Path
+        The path to the test directory.
+    elements : str
+        The parametrized elements to write to the element file.
+
+    """
+    # Create test file
+    test_file_path = tmp_path / "elements.txt"
+    with open(test_file_path, "w") as f:
+        f.write(elements)
+
+    # Get test config
+    infile = Path(__file__).parent / "data" / "gmd_mean.yaml"
+    # Read test config
+    contents = yaml.load(infile)
+    # Working directory
+    workdir = tmp_path / "workdir"
+    contents["workdir"] = str(workdir.resolve())
+    # Output directory
+    outdir = tmp_path / "outdir"
+    # Storage
+    contents["storage"]["uri"] = str(outdir.resolve())
+    # Write new test config
+    outfile = tmp_path / "in.yaml"
+    yaml.dump(contents, stream=outfile)
+    # Run command arguments
+    run_args = [
+        str(outfile.absolute()),
+        "--verbose",
+        "debug",
+        "--element",
+        str(test_file_path.resolve()),
+    ]
+    # Invoke run command
+    run_result = runner.invoke(run, run_args)
+    # Check
+    assert run_result.exit_code == 0
+
+
+@pytest.mark.parametrize(
+    "elements, expected_list",
+    [
+        ("sub-01,ses-01", [("sub-01", "ses-01")]),
+        (
+            "sub-01,ses-01\nsub-02,ses-01",
+            [("sub-01", "ses-01"), ("sub-02", "ses-01")],
+        ),
+        ("sub-01, ses-01", [("sub-01", "ses-01")]),
+        (
+            "sub-01, ses-01\nsub-02, ses-01",
+            [("sub-01", "ses-01"), ("sub-02", "ses-01")],
+        ),
+        (" sub-01 , ses-01 ", [("sub-01", "ses-01")]),
+        (
+            " sub-01 , ses-01 \n sub-02, ses-01 ",
+            [("sub-01", "ses-01"), ("sub-02", "ses-01")],
+        ),
+    ],
+)
+def test_multi_element_access(
+    tmp_path: Path, elements: str, expected_list: List[Tuple[str, ...]]
+) -> None:
+    """Test mulit-element parsing.
+
+    Parameters
+    ----------
+    tmp_path : pathlib.Path
+        The path to the test directory.
+    elements : str
+        The parametrized elements to write to the element file.
+    expected_list : list of tuple of str
+        The parametrized list of element tuples to expect.
+
+    """
+    # Create test file
+    test_file_path = tmp_path / "elements_multi.txt"
+    with open(test_file_path, "w") as f:
+        f.write(elements)
+    # Load element file
+    read_elements = _parse_elements_file(test_file_path)
+    # Check
+    assert read_elements == expected_list
+
+
 def test_wtf_short() -> None:
     """Test short version of wtf command."""
     # Invoke wtf command

junifer/api/tests/test_functions.py

@@ -120,7 +120,7 @@ def test_run_single_element_with_preprocessing(tmp_path: Path) -> None:
         preprocessor={
             "kind": "fMRIPrepConfoundRemover",
         },
-        elements=["sub-001"],
+        elements=["sub-01"],
     )
     # Check files
     files = list(outdir.glob("*.sqlite"))

junifer/datagrabber/base.py

@@ -58,12 +58,12 @@ class BaseDataGrabber(ABC, UpdateMetaMixin):
         """
         yield from self.get_elements()
 
-    def __getitem__(self, element: Union[str, Tuple]) -> Dict[str, Dict]:
+    def __getitem__(self, element: Union[str, Tuple[str]]) -> Dict[str, Dict]:
         """Enable indexing support.
 
         Parameters
         ----------
-        element : str or tuple
+        element : str or tuple of str
             The element to be indexed.
 
         Returns
@@ -117,6 +117,51 @@ class BaseDataGrabber(ABC, UpdateMetaMixin):
         """
         return self._datadir
 
+    def filter(self, selection: List[Union[str, Tuple[str]]]) -> Iterator:
+        """Filter elements to be grabbed.
+
+        Parameters
+        ----------
+        selection : list of str or tuple
+            The list of partial element key values to filter using.
+
+        Yields
+        ------
+        object
+            An element that can be indexed by the DataGrabber.
+
+        """
+
+        def filter_func(element: Union[str, Tuple[str]]) -> bool:
+            """Filter element based on selection.
+
+            Parameters
+            ----------
+            element : str or tuple of str
+                The element to be filtered.
+
+            Returns
+            -------
+            bool
+                If the element passes the filter or not.
+
+            """
+            # Convert element to tuple
+            if not isinstance(element, tuple):
+                element = (element,)
+            # Filter based on selection kind
+            if isinstance(selection[0], str):
+                for opt in selection:
+                    if opt in element:
+                        return True
+            elif isinstance(selection[0], tuple):
+                for opt in selection:
+                    if set(opt).issubset(element):
+                        return True
+            return False
+
+        yield from filter(filter_func, self.get_elements())
+
     @abstractmethod
     def get_element_keys(self) -> List[str]:
         """Get element keys.
@@ -136,7 +181,7 @@ class BaseDataGrabber(ABC, UpdateMetaMixin):
         )
 
     @abstractmethod
-    def get_elements(self) -> List:
+    def get_elements(self) -> List[Union[str, Tuple[str]]]:
         """Get elements.
 
         Returns

junifer/datagrabber/tests/test_base.py

@@ -67,3 +67,60 @@ def test_BaseDataGrabber() -> None:
 
     with pytest.raises(NotImplementedError):
         dg.get_item(subject=1)  # type: ignore
+
+
+def test_BaseDataGrabber_filter_single() -> None:
+    """Test single-keyed element filter for BaseDataGrabber."""
+
+    # Create concrete class
+    class FilterDataGrabber(BaseDataGrabber):
+        def get_item(self, subject):
+            return {"BOLD": {}}
+
+        def get_elements(self):
+            return ["sub01", "sub02", "sub03"]
+
+        def get_element_keys(self):
+            return ["subject"]
+
+    dg = FilterDataGrabber(datadir="/tmp", types=["BOLD"])
+    with dg:
+        assert "sub01" in list(dg.filter(["sub01"]))
+        assert "sub02" not in list(dg.filter(["sub01"]))
+
+
+def test_BaseDataGrabber_filter_multi() -> None:
+    """Test multi-keyed element filter for BaseDataGrabber."""
+
+    # Create concrete class
+    class FilterDataGrabber(BaseDataGrabber):
+        def get_item(self, subject):
+            return {"BOLD": {}}
+
+        def get_elements(self):
+            return [
+                ("sub01", "rest"),
+                ("sub01", "movie"),
+                ("sub02", "rest"),
+                ("sub02", "movie"),
+                ("sub03", "rest"),
+                ("sub03", "movie"),
+            ]
+
+        def get_element_keys(self):
+            return ["subject", "task"]
+
+    dg = FilterDataGrabber(datadir="/tmp", types=["BOLD"])
+    with dg:
+        assert ("sub01", "rest") in list(
+            dg.filter([("sub01", "rest")])  # type: ignore
+        )
+        assert ("sub01", "movie") not in list(
+            dg.filter([("sub01", "rest")])  # type: ignore
+        )
+        assert ("sub02", "rest") not in list(
+            dg.filter([("sub01", "rest")])  # type: ignore
+        )
+        assert ("sub02", "movie") not in list(
+            dg.filter([("sub01", "rest")])  # type: ignore
+        )