cursor

Author: singlesp

cubids/cubids.py

@@ -123,7 +123,9 @@ def __init__(
         self.data_dict = {}  # data dictionary for TSV outputs
         self.use_datalad = use_datalad  # True if flag set, False if flag unset
         self.schema = load_schema(schema_json)
-        self.is_longitudinal = self._infer_longitudinal()  # inferred from dataset structure
+        self.is_longitudinal = (
+            self._infer_longitudinal()
+        )  # inferred from dataset structure
 
         if self.use_datalad:
             self.init_datalad()
@@ -186,7 +188,9 @@ def reset_bids_layout(self, validate=False):
             re.compile(r"/\."),
         ]
 
-        indexer = bids.BIDSLayoutIndexer(validate=validate, ignore=ignores, index_metadata=False)
+        indexer = bids.BIDSLayoutIndexer(
+            validate=validate, ignore=ignores, index_metadata=False
+        )
 
         self._layout = bids.BIDSLayout(self.path, validate=validate, indexer=indexer)
 
@@ -293,7 +297,9 @@ def datalad_undo_last_commit(self):
             If there are untracked changes in the datalad dataset.
         """
         if not self.is_datalad_clean():
-            raise Exception("Untracked changes present. Run clear_untracked_changes first")
+            raise Exception(
+                "Untracked changes present. Run clear_untracked_changes first"
+            )
         reset_proc = subprocess.run(["git", "reset", "--hard", "HEAD~1"], cwd=self.path)
         reset_proc.check_returncode()
 
@@ -417,7 +423,9 @@ def add_file_collections(self):
                 continue
 
             # Add file collection metadata to the sidecar
-            files, collection_metadata = utils.collect_file_collections(self.layout, path)
+            files, collection_metadata = utils.collect_file_collections(
+                self.layout, path
+            )
             filepaths = [f.path for f in files]
             checked_files.extend(filepaths)
 
@@ -439,7 +447,9 @@ def add_file_collections(self):
 
         self.reset_bids_layout()
 
-    def apply_tsv_changes(self, summary_tsv, files_tsv, new_prefix, raise_on_error=True):
+    def apply_tsv_changes(
+        self, summary_tsv, files_tsv, new_prefix, raise_on_error=True
+    ):
         """Apply changes documented in the edited summary tsv and generate the new tsv files.
 
         This function looks at the RenameEntitySet and MergeInto
@@ -475,11 +485,15 @@ def apply_tsv_changes(self, summary_tsv, files_tsv, new_prefix, raise_on_error=T
         files_df = pd.read_table(files_tsv)
 
         # Check that the MergeInto column only contains valid merges
-        ok_merges, deletions = check_merging_operations(summary_tsv, raise_on_error=raise_on_error)
+        ok_merges, deletions = check_merging_operations(
+            summary_tsv, raise_on_error=raise_on_error
+        )
 
         merge_commands = []
         for source_id, dest_id in ok_merges:
-            dest_files = files_df.loc[(files_df[["ParamGroup", "EntitySet"]] == dest_id).all(1)]
+            dest_files = files_df.loc[
+                (files_df[["ParamGroup", "EntitySet"]] == dest_id).all(1)
+            ]
             source_files = files_df.loc[
                 (files_df[["ParamGroup", "EntitySet"]] == source_id).all(1)
             ]
@@ -490,12 +504,16 @@ def apply_tsv_changes(self, summary_tsv, files_tsv, new_prefix, raise_on_error=T
             for dest_nii in dest_files.FilePath:
                 dest_json = utils.img_to_new_ext(self.path + dest_nii, ".json")
                 if Path(dest_json).exists() and Path(source_json).exists():
-                    merge_commands.append(f"cubids bids-sidecar-merge {source_json} {dest_json}")
+                    merge_commands.append(
+                        f"cubids bids-sidecar-merge {source_json} {dest_json}"
+                    )
 
         # Get the delete commands
         to_remove = []
         for rm_id in deletions:
-            files_to_rm = files_df.loc[(files_df[["ParamGroup", "EntitySet"]] == rm_id).all(1)]
+            files_to_rm = files_df.loc[
+                (files_df[["ParamGroup", "EntitySet"]] == rm_id).all(1)
+            ]
 
             for rm_me in files_to_rm.FilePath:
                 if Path(self.path + rm_me).exists():
@@ -692,19 +710,10 @@ def change_filename(self, filepath, entities):
                 new_context = new_path.replace(new_scan_end, "_aslcontext.tsv")
                 self.new_filenames.append(new_context)
 
-            old_m0scan = filepath.replace(scan_end, "_m0scan.nii.gz")
-            if Path(old_m0scan).exists():
-                self.old_filenames.append(old_m0scan)
-                new_scan_end = "_" + suffix + old_ext
-                new_m0scan = new_path.replace(new_scan_end, "_m0scan.nii.gz")
-                self.new_filenames.append(new_m0scan)
-
-            old_mjson = filepath.replace(scan_end, "_m0scan.json")
-            if Path(old_mjson).exists():
-                self.old_filenames.append(old_mjson)
-                new_scan_end = "_" + suffix + old_ext
-                new_mjson = new_path.replace(new_scan_end, "_m0scan.json")
-                self.new_filenames.append(new_mjson)
+            # Do NOT rename M0 scans or their JSON sidecars. M0 files should
+            # retain their original filenames to preserve independent variability.
+            # The IntendedFor field in M0 JSONs will be updated below to point
+            # to the newly renamed ASL files.
 
             old_labeling = filepath.replace(scan_end, "_asllabeling.jpg")
             if Path(old_labeling).exists():
@@ -739,13 +748,17 @@ def change_filename(self, filepath, entities):
                         # remove old filename
                         data["IntendedFor"].remove(item)
                         # add new filename
-                        data["IntendedFor"].append(utils._get_participant_relative_path(new_path))
+                        data["IntendedFor"].append(
+                            utils._get_participant_relative_path(new_path)
+                        )
 
                     if item == utils._get_bidsuri(filepath, self.path):
                         # remove old filename
                         data["IntendedFor"].remove(item)
                         # add new filename
-                        data["IntendedFor"].append(utils._get_bidsuri(new_path, self.path))
+                        data["IntendedFor"].append(
+                            utils._get_bidsuri(new_path, self.path)
+                        )
 
                 # update the json with the new data dictionary
                 utils._update_json(filename_with_if, data)
@@ -922,7 +935,9 @@ def _purge_associations(self, scans):
 
                 if "/func/" in str(path):
                     # add tsvs
-                    tsv = utils.img_to_new_ext(str(path), ".tsv").replace("_bold", "_events")
+                    tsv = utils.img_to_new_ext(str(path), ".tsv").replace(
+                        "_bold", "_events"
+                    )
                     if Path(tsv).exists():
                         to_remove.append(tsv)
                     # add tsv json (if exists)
@@ -1277,17 +1292,23 @@ def get_param_groups_dataframes(self):
             long_name = big_df.loc[row, "FilePath"]
             big_df.loc[row, "FilePath"] = long_name.replace(self.path, "")
 
-        summary = utils._order_columns(pd.concat(param_group_summaries, ignore_index=True))
+        summary = utils._order_columns(
+            pd.concat(param_group_summaries, ignore_index=True)
+        )
 
         # create new col that strings key and param group together
-        summary["KeyParamGroup"] = summary["EntitySet"] + "__" + summary["ParamGroup"].map(str)
+        summary["KeyParamGroup"] = (
+            summary["EntitySet"] + "__" + summary["ParamGroup"].map(str)
+        )
 
         # move this column to the front of the dataframe
         key_param_col = summary.pop("KeyParamGroup")
         summary.insert(0, "KeyParamGroup", key_param_col)
 
         # do the same for the files df
-        big_df["KeyParamGroup"] = big_df["EntitySet"] + "__" + big_df["ParamGroup"].map(str)
+        big_df["KeyParamGroup"] = (
+            big_df["EntitySet"] + "__" + big_df["ParamGroup"].map(str)
+        )
 
         # move this column to the front of the dataframe
         key_param_col = big_df.pop("KeyParamGroup")
@@ -1362,8 +1383,12 @@ def get_tsvs(self, path_prefix):
 
         big_df, summary = self.get_param_groups_dataframes()
 
-        summary = summary.sort_values(by=["Modality", "EntitySetCount"], ascending=[True, False])
-        big_df = big_df.sort_values(by=["Modality", "EntitySetCount"], ascending=[True, False])
+        summary = summary.sort_values(
+            by=["Modality", "EntitySetCount"], ascending=[True, False]
+        )
+        big_df = big_df.sort_values(
+            by=["Modality", "EntitySetCount"], ascending=[True, False]
+        )
 
         # Create json dictionaries for summary and files tsvs
         self.create_data_dictionary()

cubids/tests/test_perf_m0.py

@@ -0,0 +1,82 @@
+"""Tests for ASL/M0 renaming behavior.
+
+Ensures that when ASL scans are renamed with variant acquisition labels:
+- aslcontext files are renamed to match the ASL scan
+- M0 files (nii/json) are NOT renamed
+- M0 JSON IntendedFor entries are updated to point to the new ASL path
+"""
+
+import json
+from pathlib import Path
+
+from cubids.cubids import CuBIDS
+
+
+def _write(path: Path, content: str = ""):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(content)
+
+
+def test_m0_not_renamed_but_aslcontext_is_and_intendedfor_updated(tmp_path):
+    bids_root = tmp_path / "bids"
+    sub = "sub-01"
+    ses = "ses-01"
+
+    # Create minimal perf files
+    perf_dir = bids_root / sub / ses / "perf"
+    perf_dir.mkdir(parents=True, exist_ok=True)
+
+    asl_base = perf_dir / f"{sub}_{ses}_asl.nii.gz"
+    asl_json = perf_dir / f"{sub}_{ses}_asl.json"
+    m0_base = perf_dir / f"{sub}_{ses}_m0scan.nii.gz"
+    m0_json = perf_dir / f"{sub}_{ses}_m0scan.json"
+    aslcontext = perf_dir / f"{sub}_{ses}_aslcontext.tsv"
+
+    # Touch NIfTIs (empty is fine for this test) and sidecars
+    asl_base.write_bytes(b"")
+    m0_base.write_bytes(b"")
+
+    _write(asl_json, json.dumps({}))
+
+    # M0 IntendedFor should reference the ASL time series (participant-relative path)
+    intended_for_rel = f"{ses}/perf/{sub}_{ses}_asl.nii.gz"
+    _write(m0_json, json.dumps({"IntendedFor": [intended_for_rel]}))
+
+    _write(aslcontext, "label\ncontrol\nlabel\ncontrol\n")
+
+    c = CuBIDS(str(bids_root))
+
+    # Rename the ASL scan by adding a variant acquisition
+    entities = {"suffix": "asl", "acquisition": "VARIANTTest"}
+    c.change_filename(str(asl_base), entities)
+
+    # Old/new filenames prepared for ASL and aslcontext, but NOT for M0
+    assert str(asl_base) in c.old_filenames
+    assert any(fn.endswith("_asl.json") for fn in c.old_filenames)
+    assert any(fn.endswith("_aslcontext.tsv") for fn in c.old_filenames)
+
+    assert not any(fn.endswith("_m0scan.nii.gz") for fn in c.old_filenames)
+    assert not any(fn.endswith("_m0scan.json") for fn in c.old_filenames)
+
+    # Compute expected new ASL path and aslcontext path
+    expected_new_asl = perf_dir / f"{sub}_{ses}_acq-VARIANTTest_asl.nii.gz"
+    expected_new_aslcontext = perf_dir / f"{sub}_{ses}_acq-VARIANTTest_aslcontext.tsv"
+
+    assert str(expected_new_asl) in c.new_filenames
+    assert str(expected_new_aslcontext) in c.new_filenames
+
+    # M0 files remain with original names
+    assert m0_base.exists()
+    assert m0_json.exists()
+
+    # But M0 IntendedFor should now point to the new ASL relative path
+    with open(m0_json, "r") as f:
+        m0_meta = json.load(f)
+
+    new_rel = f"{ses}/perf/{sub}_{ses}_acq-VARIANTTest_asl.nii.gz"
+    assert "IntendedFor" in m0_meta
+    assert new_rel in m0_meta["IntendedFor"]
+    # Ensure old reference removed
+    assert intended_for_rel not in m0_meta["IntendedFor"]
+
+