[ENH]: Unix path expansion directives support for PatternDataGrabber #345

Merged
synchon merged 4 commits from feat/unix-path-expansion-pattern-dg into main 2024-05-22 17:18:14 +00:00
4 changed files with 110 additions and 6 deletions

View file

@ -0,0 +1 @@
Allow Unix path expansion directives to be used in :class:`.PatternDataGrabber` ``patterns`` by `Synchon Mandal`_

View file

@ -314,6 +314,53 @@ This approach can be used directly from the YAML, like so:
uri: "https://gin.g-node.org/juaml/datalad-example-bids"
rootdir: example_bids_ses
Advanced: Using Unix-like path expansion directives
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
It is also possible to use some advanced Unix-like path expansion tricks to
define our patterns.
A very common thing would be to use ``*`` to match any number of
characters but we cannot use it right after a replacement like:
.. code-block:: python
"derivatives/freesurfer/{subject}*"
or if there are multiple files or no files which can be globbed.
We can also use ``[]`` and ``[!]`` to glob certain tricky files like with the
case of FreeSurfer derivatives. The file structure seen in a typical
FreeSurfer derivative of a dataset (like ``AOMIC`` ones) is like so:
.. code-block::
.
└── derivatives
└── freesurfer
├── fsaverage
│ ├── mri
│ | ├── T1.mgz
│ | └── ...
│ └── ...
├── sub-01
│ ├── mri
│ | ├── T1.mgz
│ | └── ...
│ | └── ...
│ └── ...
...
With a structure like this, it would be cumbersome to write custom methods
for the class and thus we could use a pattern like this:
.. code-block:: python
"derivatives/freesurfer/[!f]{subject}/mri/T1.mg[z]"
This would ignore the ``fsaverage`` directory as a subject and let ``T1.mgz`` be
fetched as there can be many files with the same prefix.
.. _extending_datagrabbers_base:
Option B: Extending from BaseDataGrabber

View file

@ -214,18 +214,25 @@ class PatternDataGrabber(BaseDataGrabber):
t_replacements = [
x for x in self.replacements if f"{{{x}}}" in pattern
]
# Ops on re_pattern
# Remove negated unix glob pattern i.e., [!...] for re_pattern
re_pattern = re.sub(r"\[!.?\]", "", re_pattern)
# Remove enclosing square brackets from unix glob pattern i.e., [...]
# for re_pattern
re_pattern = re.sub(r"\[|\]", "", re_pattern)
# Iteratively replace the first of each with a named group definition
for t_r in t_replacements:
# Replace the first of each with a named group definition
re_pattern = re_pattern.replace(f"{{{t_r}}}", f"(?P<{t_r}>.*)", 1)
# Iteratively replace the second appearance of each with the named
# group back reference
for t_r in t_replacements:
# Replace the second appearance of each with the named group
# back reference
re_pattern = re_pattern.replace(f"{{{t_r}}}", f"(?P={t_r})")
# Ops on glob_pattern
# Iteratively replace replacements with wildcard i.e., *
# for glob_pattern
for t_r in t_replacements:
glob_pattern = glob_pattern.replace(f"{{{t_r}}}", "*")
return re_pattern, glob_pattern, t_replacements
def _replace_patterns_glob(self, element: Dict, pattern: str) -> str:
@ -254,6 +261,10 @@ class PatternDataGrabber(BaseDataGrabber):
f"The element keys must be {self.replacements}, "
f"element has {list(element.keys())}."
)
# Remove negated unix glob pattern i.e., [!...]
pattern = re.sub(r"\[!.?\]", "", pattern)
# Remove enclosing square brackets from unix glob pattern i.e., [...]
pattern = re.sub(r"\[|\]", "", pattern)
return pattern.format(**element)
def _get_path_from_patterns(

View file

@ -236,6 +236,51 @@ def test_PatternDataGrabber(tmp_path: Path) -> None:
assert out1["VBM_GM"]["path"] != out2["VBM_GM"]["path"]
def test_PatternDataGrabber_unix_path_expansion(tmp_path: Path) -> None:
"""Test PatterDataGrabber for patterns with unix path expansion.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
# Create test data root dir
freesurfer_dir = tmp_path / "derivatives" / "freesurfer"
freesurfer_dir.mkdir(parents=True, exist_ok=True)
# Create test data sub dirs and files
for dir_name in ["fsaverage", "sub-0001"]:
mri_dir = freesurfer_dir / dir_name / "mri"
mri_dir.mkdir(parents=True, exist_ok=True)
# Create files
(mri_dir / "T1.mgz").touch(exist_ok=True)
(mri_dir / "aseg.mgz").touch(exist_ok=True)
# Create datagrabber
dg = PatternDataGrabber(
datadir=tmp_path,
types=["FreeSurfer"],
patterns={
"FreeSurfer": {
"pattern": "derivatives/freesurfer/[!f]{subject}/mri/T1.mg[z]",
"aseg": {
"pattern": (
"derivatives/freesurfer/[!f]{subject}/mri/aseg.mg[z]"
)
},
},
},
replacements=["subject"],
)
# Check that "fsaverage" is filtered
elements = dg.get_elements()
assert elements == ["sub-0001"]
# Fetch data
out = dg["sub-0001"]
# Check paths are found
assert set(out["FreeSurfer"].keys()) == {"path", "aseg", "meta"}
assert list(out["FreeSurfer"]["aseg"].keys()) == ["path"]
def test_PatternDataGrabber_confounds_format_error_on_init() -> None:
"""Test PatterDataGrabber confounds format error on initialisation."""
with pytest.raises(