[ENH] use datalad copy files

cohort_creator/_utils.py

@@ -120,25 +120,6 @@ def filter_excluded_participants(pth: Path, participants: list[str] | None) -> N
     participants_df.to_csv(participants_tsv, sep="\t", index=False)
 
 
-def copy_top_files(src_pth: Path, target_pth: Path, datatypes: list[str]) -> None:
-    """Copy top files from BIDS src_pth to BIDS target_pth."""
-    top_files = ["dataset_description.json", "participants.*", "README*"]
-    if "func" in datatypes:
-        top_files.extend(["*task-*_events.tsv", "*task-*_events.json", "*task-*_bold.json"])
-    if "anat" in datatypes:
-        top_files.append("*T1w.json")
-
-    for top_file_ in top_files:
-        for f in src_pth.glob(top_file_):
-            if (target_pth / f.name).exists():
-                cc_log.debug(f"      file already present:\n       '{(target_pth / f.name)}'")
-                continue
-            try:
-                shutil.copy(src=f, dst=target_pth, follow_symlinks=True)
-            except FileNotFoundError:
-                cc_log.error(f"      Could not find file '{f}'")
-
-
 def check_tsv_content(tsv_file: Path | str) -> pd.DataFrame:
     tsv_file = Path(tsv_file).resolve()
     if not tsv_file.exists():

cohort_creator/copy_files.py

@@ -0,0 +1,48 @@
+"""Module to handle copying data out of source datalad datasets."""
+
+from __future__ import annotations
+
+import shutil
+from pathlib import Path
+
+import pandas as pd
+
+from cohort_creator.logger import cc_logger
+
+# from datalad import api
+# from datalad.support.exceptions import IncompleteResultsError
+
+
+cc_log = cc_logger()
+
+
+def copy_files(
+    output_dir: Path,
+    datasets: pd.DataFrame,
+    participants: pd.DataFrame | None,
+    dataset_types: list[str],
+    datatypes: list[str],
+    task: str,
+    space: str,
+    bids_filter: None | dict[str, dict[str, dict[str, str]]] = None,
+):
+    pass
+
+
+def copy_top_files(src_pth: Path, target_pth: Path, datatypes: list[str]) -> None:
+    """Copy top files from BIDS src_pth to BIDS target_pth."""
+    top_files = ["dataset_description.json", "participants.*", "README*"]
+    if "func" in datatypes:
+        top_files.extend(["*task-*_events.tsv", "*task-*_events.json", "*task-*_bold.json"])
+    if "anat" in datatypes:
+        top_files.append("*T1w.json")
+
+    for top_file_ in top_files:
+        for f in src_pth.glob(top_file_):
+            if (target_pth / f.name).exists():
+                cc_log.debug(f"      file already present:\n       '{(target_pth / f.name)}'")
+                continue
+            try:
+                shutil.copy(src=f, dst=target_pth, follow_symlinks=True)
+            except FileNotFoundError:
+                cc_log.error(f"      Could not find file '{f}'")

cohort_creator/main.py

@@ -18,7 +18,6 @@
 
 from cohort_creator._utils import (
     add_study_tsv,
-    copy_top_files,
     create_ds_description,
     create_tsv_participant_session_in_datasets,
     dataset_path,
@@ -42,6 +41,7 @@
     sourcedata,
 )
 from cohort_creator.bagelify import bagelify, new_bagel
+from cohort_creator.copy_files import copy_top_files
 from cohort_creator.data.utils import is_known_dataset
 from cohort_creator.logger import cc_logger
 

tests/conftest.py

@@ -5,6 +5,8 @@
 import pytest
 from datalad import api
 
+from cohort_creator._cli import create_yoda
+
 
 def root_dir():
     return Path(__file__).parent.parent
@@ -19,6 +21,12 @@ def bids_examples():
     return path_test_data() / "bids-examples"
 
 
+@pytest.fixture
+def output_dir(tmp_path):
+    create_yoda(output_dir=tmp_path)
+    return tmp_path
+
+
 @pytest.fixture
 def install_dataset():
     def _install_dataset(dataset_name: str):

tests/test_cohort_creator.py

@@ -3,19 +3,11 @@
 from __future__ import annotations
 
 import pandas as pd
-import pytest
 
-from cohort_creator._cli import create_yoda
 from cohort_creator._utils import sourcedata
 from cohort_creator.main import construct_cohort, get_data, install_datasets
 
 
-@pytest.fixture
-def output_dir(tmp_path):
-    create_yoda(output_dir=tmp_path)
-    return tmp_path
-
-
 def test_install_datasets(output_dir, caplog):
     install_datasets(
         datasets=["ds000001", "foo"],

tests/test_copy_files.py

@@ -0,0 +1,47 @@
+"""Utilities."""
+
+from __future__ import annotations
+
+import pandas as pd
+
+from cohort_creator.copy_files import copy_files
+from cohort_creator.main import install_datasets
+
+
+def test_copy_top_files(output_dir):
+    dataset_types = ["raw"]
+    # datatypes = ["anat"]
+    install_datasets(
+        datasets=["ds000001", "foo"], output_dir=output_dir, dataset_types=dataset_types
+    )
+    # copy_top_files(src_pth=src_pth, target_pth=target_pth, datatypes=datatypes)
+
+
+def test_copy_files(output_dir):
+    participants = pd.DataFrame(
+        {"DatasetID": ["ds000001"], "SubjectID": ["sub-01"], "SessionID": [""]}
+    )
+    datasets = pd.DataFrame(
+        {
+            "DatasetID": ["ds000001"],
+            "PortalURI": ["https://github.com/OpenNeuroDatasets-JSONLD/ds000001.git"],
+        }
+    )
+    dataset_types = ["raw"]
+    datatypes = ["anat"]
+    install_datasets(
+        datasets=["ds000001", "foo"], output_dir=output_dir, dataset_types=dataset_types
+    )
+    copy_files(
+        output_dir=output_dir,
+        datasets=datasets,
+        participants=participants,
+        dataset_types=dataset_types,
+        datatypes=datatypes,
+        space="not_used_for_raw",
+        task="*",
+    )
+
+    assert (
+        output_dir / "study-ds000001" / "bids" / "sub-01" / "anat" / "sub-03_T1w.nii.gzz"
+    ).exists()