🐛 Restore header to censors.tsv

CPAC/nuisance/nuisance.py

@@ -1,4 +1,4 @@
-# Copyright (C) 2012-2024  C-PAC Developers
+# Copyright (C) 2012-2025  C-PAC Developers
 
 # This file is part of C-PAC.
 
@@ -14,6 +14,8 @@
 
 # You should have received a copy of the GNU Lesser General Public
 # License along with C-PAC. If not, see <https://www.gnu.org/licenses/>.
+"""Nusiance regression."""
+
 import os
 from typing import Literal
 
@@ -29,6 +31,7 @@
 from CPAC.nuisance.utils import (
     find_offending_time_points,
     generate_summarize_tissue_mask,
+    load_censor_tsv,
     temporal_variance_mask,
 )
 from CPAC.nuisance.utils.compcor import (
@@ -302,7 +305,7 @@
             raise ValueError(msg)
 
         try:
-            regressors = np.loadtxt(regressor_file)
+            regressors = load_censor_tsv(regressor_file, regressor_length)
         except (OSError, TypeError, UnicodeDecodeError, ValueError) as error:
             msg = f"Could not read regressor {regressor_type} from {regressor_file}."
             raise OSError(msg) from error
@@ -382,7 +385,7 @@
     if custom_file_paths:
         for custom_file_path in custom_file_paths:
             try:
-                custom_regressor = np.loadtxt(custom_file_path)
+                custom_regressor = load_censor_tsv(custom_file_path, regressor_length)
             except:
                 msg = "Could not read regressor {0} from {1}.".format(
                     "Custom", custom_file_path
@@ -421,7 +424,7 @@
             censor_volumes = np.ones((regressor_length,), dtype=int)
         else:
             try:
-                censor_volumes = np.loadtxt(regressor_file)
+                censor_volumes = load_censor_tsv(regressor_file, regressor_length)
             except:
                 msg = (
                     f"Could not read regressor {regressor_type} from {regressor_file}."

CPAC/nuisance/tests/test_utils.py

@@ -1,33 +1,55 @@
+# Copyright (C) 2019-2025  C-PAC Developers
+
+# This file is part of C-PAC.
+
+# C-PAC is free software: you can redistribute it and/or modify it under
+# the terms of the GNU Lesser General Public License as published by the
+# Free Software Foundation, either version 3 of the License, or (at your
+# option) any later version.
+
+# C-PAC is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
+# License for more details.
+
+# You should have received a copy of the GNU Lesser General Public
+# License along with C-PAC. If not, see <https://www.gnu.org/licenses/>.
+"""Test nuisance utilities."""
+
+from importlib.resources import as_file, files
 import os
+from pathlib import Path
+from random import randint
 import tempfile
 
 import numpy as np
-import pkg_resources as p
 import pytest
 
-from CPAC.nuisance.utils import calc_compcor_components, find_offending_time_points
+from CPAC.nuisance.utils import (
+    calc_compcor_components,
+    find_offending_time_points,
+    load_censor_tsv,
+)
 from CPAC.utils.monitoring.custom_logging import getLogger
 
 logger = getLogger("CPAC.nuisance.tests")
 
-mocked_outputs = p.resource_filename(
-    "CPAC", os.path.join("nuisance", "tests", "motion_statistics")
-)
+_mocked_outputs = files("CPAC").joinpath("nuisance/tests/motion_statistics")
 
 
 @pytest.mark.skip(reason="needs refactoring")
 def test_find_offending_time_points():
     dl_dir = tempfile.mkdtemp()
     os.chdir(dl_dir)
-
-    censored = find_offending_time_points(
-        os.path.join(mocked_outputs, "FD_J.1D"),
-        os.path.join(mocked_outputs, "FD_P.1D"),
-        os.path.join(mocked_outputs, "DVARS.1D"),
-        2.0,
-        2.0,
-        "1.5SD",
-    )
+    with as_file(_mocked_outputs) as mocked_outputs:
+        censored = find_offending_time_points(
+            str(mocked_outputs / "FD_J.1D"),
+            str(mocked_outputs / "FD_P.1D"),
+            str(mocked_outputs / "DVARS.1D"),
+            2.0,
+            2.0,
+            "1.5SD",
+        )
 
     censored = np.loadtxt(censored).astype(bool)
 
@@ -41,4 +63,21 @@
 
     compcor_filename = calc_compcor_components(data_filename, 5, mask_filename)
     logger.info("compcor components written to %s", compcor_filename)
-    assert 0 == 1
+
+
+@pytest.mark.parametrize("header", [True, False])
+def test_load_censor_tsv(header: bool, tmp_path: Path) -> None:
+    """Test loading of censor tsv files with and without headers."""
+    expected_length = 3
+    filepath = tmp_path / "censor.tsv"
+    with filepath.open("w") as f:
+        if header:
+            f.write("censor\n")
+        for i in range(expected_length):
+            f.write(f"{randint(0, 1)}\n")
+    censors = load_censor_tsv(str(filepath), expected_length)
+    assert (
+        censors.shape[0] == expected_length
+    ), "Length of censors does not match expected length"
+    with pytest.raises(ValueError, match="expected length"):
+        load_censor_tsv(str(filepath), expected_length + 1)

CPAC/nuisance/utils/__init__.py

@@ -1,4 +1,4 @@
-# Copyright (C) 2019-2024  C-PAC Developers
+# Copyright (C) 2019-2025  C-PAC Developers
 
 # This file is part of C-PAC.
 
@@ -21,6 +21,7 @@
 from .utils import (
     find_offending_time_points,
     generate_summarize_tissue_mask,
+    load_censor_tsv,
     NuisanceRegressor,
     temporal_variance_mask,
 )
@@ -30,6 +31,7 @@
     "compcor",
     "find_offending_time_points",
     "generate_summarize_tissue_mask",
+    "load_censor_tsv",
     "NuisanceRegressor",
     "temporal_variance_mask",
 ]

CPAC/nuisance/utils/utils.py

@@ -1,4 +1,4 @@
-# Copyright (C) 2019-2024  C-PAC Developers
+# Copyright (C) 2019-2025  C-PAC Developers
 
 # This file is part of C-PAC.
 
@@ -21,6 +21,7 @@
 import re
 from typing import Optional
 
+import numpy as np
 from nipype.interfaces import afni, ants, fsl
 import nipype.interfaces.utility as util
 from nipype.pipeline.engine import Workflow
@@ -139,7 +140,7 @@
     censor_vector[extended_censors] = 0
 
     out_file_path = os.path.join(os.getcwd(), "censors.tsv")
-    np.savetxt(out_file_path, censor_vector, fmt="%d", comments="")
+    np.savetxt(out_file_path, censor_vector, fmt="%d", header="censor", comments="")
 
     return out_file_path
 
@@ -860,3 +861,19 @@
     def __repr__(self) -> str:
         """Return a string representation of the nuisance regressor."""
         return NuisanceRegressor.encode(self.selector)
+
+
+def load_censor_tsv(filepath: str, expected_length: int) -> np.ndarray:
+    """Load censor TSV and verify length."""
+    header = False
+    censor = np.empty((0))
+    try:
+        censor = np.loadtxt(filepath)
+    except ValueError:
+        header = True
+    if header or censor.shape[0] == expected_length + 1:
+        censor = np.loadtxt(filepath, skiprows=1)
+    if censor.shape[0] == expected_length:
+        return censor
+    msg = f"Censor file length ({censor.shape[0]}) does not match expected length ({expected_length})."
+    raise ValueError(msg)