♻️ Skip header when loading TSV into NumPy

Ref #2228 (comment), #2228 (comment)

Co-authored-by: Biraj Shrestha  <[email protected]>

Author: shnizzedy

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 @@ def gather_nuisance(
             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 @@ def gather_nuisance(
     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 @@ def gather_nuisance(
             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

@@ -18,12 +18,18 @@
 
 from importlib.resources import as_file, files
 import os
+from pathlib import Path
+from random import randint
 import tempfile
 
 import numpy as np
 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")
@@ -58,3 +64,20 @@ def test_calc_compcor_components():
     compcor_filename = calc_compcor_components(data_filename, 5, mask_filename)
     logger.info("compcor components written to %s", compcor_filename)
 
+
+@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
@@ -860,3 +861,19 @@ def encode(selector: dict) -> str:
     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)