tsv to pq & corr to connectome

jpillai00 · jpillai00 · commit a9ab56470ca8 · 2026-04-16T14:53:59.000-04:00
diff --git a/src/rbc/bids/metrics.py b/src/rbc/bids/metrics.py
@@ -89,13 +89,13 @@ def export_metrics(
             outputs.timeseries[atl],
             suffix="timeseries",
             desc="mean",
-            extension=".tsv",
+            extension=".parquet",
             atlas=bids_safe_label(atl),
         )
         mex.save(
-            outputs.correlation_matrix[atl],
-            suffix="correlations",
+            outputs.connectome[atl],
+            suffix="connectome",
             desc="pearson",
-            extension=".tsv",
+            extension=".parquet",
             atlas=bids_safe_label(atl),
         )
diff --git a/src/rbc/bids/qc.py b/src/rbc/bids/qc.py
@@ -114,6 +114,6 @@ def export_qc(
             outputs.qc_file[reg],
             suffix="quality",
             desc="xcp",
-            extension=".tsv",
+            extension=".parquet",
             extra={"reg": bids_safe_label(reg)},
         )
diff --git a/src/rbc/core/metrics/timeseries.py b/src/rbc/core/metrics/timeseries.py
@@ -11,6 +11,7 @@
 from typing import NamedTuple
 
 import numpy as np
+import polars as pl
 
 from rbc.core.niwrap import generate_exec_folder
 
@@ -113,7 +114,7 @@ class TimeseriesOutputs(NamedTuple):
     """Outputs from :func:`compute_timeseries`."""
 
     timeseries: Path
-    correlation_matrix: Path
+    connectome: Path
     labels: np.ndarray
 
 
@@ -134,7 +135,7 @@ def compute_timeseries(
             *in_file*.
 
     Returns:
-        :class:`TimeseriesOutputs` containing paths to the TSV files and
+        :class:`TimeseriesOutputs` containing paths to the Parquet files and
         the ROI labels array.
     """
     import nibabel as nib
@@ -163,14 +164,16 @@ def compute_timeseries(
     out_dir.mkdir(parents=True, exist_ok=True)
 
     stem = in_file.name.split(".nii")[0]
-    ts_path = out_dir / f"{stem}_timeseries.tsv"
-    corr_path = out_dir / f"{stem}_correlation_matrix.tsv"
+    ts_path = out_dir / f"{stem}_timeseries.parquet"
+    corr_path = out_dir / f"{stem}_connectome.parquet"
 
-    np.savetxt(ts_path, ts, delimiter="\t")
-    np.savetxt(corr_path, corr, delimiter="\t")
+    col_names = [str(label) for label in labels]
+
+    pl.DataFrame(ts, schema=col_names).write_parquet(ts_path)
+    pl.DataFrame(corr, schema=col_names).write_parquet(corr_path)
 
     return TimeseriesOutputs(
         timeseries=ts_path,
-        correlation_matrix=corr_path,
+        connectome=corr_path,
         labels=labels,
     )
diff --git a/src/rbc/core/qc/xcp.py b/src/rbc/core/qc/xcp.py
@@ -143,11 +143,12 @@ def generate_xcp_qc(
 
 
 def write_xcp_qc(metrics: XCPQCMetrics, out_path: Path) -> Path:
-    """Write XCP QC metrics as a single-row TSV file.
+    """Write XCP QC metrics as a single-row Parquet file.
 
     Args:
         metrics: A populated :class:`XCPQCMetrics` row.
         out_path: Destination file path (parent dirs created if needed).
+            Should have a ``.parquet`` extension.
 
     Returns:
         The output path (same as *out_path*).
@@ -156,7 +157,7 @@ def write_xcp_qc(metrics: XCPQCMetrics, out_path: Path) -> Path:
     out_path.parent.mkdir(parents=True, exist_ok=True)
 
     df = pl.DataFrame([metrics._asdict()])
-    df.write_csv(out_path, separator="\t")
+    df.write_parquet(out_path)
 
     return out_path
 
diff --git a/src/rbc/workflows/metrics.py b/src/rbc/workflows/metrics.py
@@ -39,8 +39,8 @@ class MetricsOutputs(NamedTuple):
         reho: Raw ReHo map.
         reho_smooth: Smoothed ReHo map.
         reho_zscored: Z-scored (smoothed) ReHo map.
-        timeseries: Atlas-based mean timeseries TSV.
-        correlation_matrix: Pairwise correlation matrix TSV.
+        timeseries: Atlas-based mean timeseries file.
+        connectome: Pairwise connectome file.
     """
 
     alff: Path
@@ -53,7 +53,7 @@ class MetricsOutputs(NamedTuple):
     reho_smooth: Path
     reho_zscored: Path
     timeseries: dict[str, Path]
-    correlation_matrix: dict[str, Path]
+    connectome: dict[str, Path]
 
 
 def single_session_metrics(
@@ -127,7 +127,5 @@ def single_session_metrics(
         reho_smooth=reho_smooth_path,
         reho_zscored=reho_zscored_path,
         timeseries={label: ts.timeseries for label, ts in ts_outputs.items()},
-        correlation_matrix={
-            label: ts.correlation_matrix for label, ts in ts_outputs.items()
-        },
+        connectome={label: ts.connectome for label, ts in ts_outputs.items()},
     )
diff --git a/src/rbc/workflows/qc.py b/src/rbc/workflows/qc.py
@@ -160,7 +160,7 @@ def single_session_qc(
         # 9. Write QC TSV
         qc_outputs.qc_file[regressor] = write_xcp_qc(
             qc_outputs.metrics[regressor],
-            work_dir / f"reg-{regressor}_qc.tsv",
+            work_dir / f"reg-{regressor}_qc.parquet",
         )
 
     # 10. RBC pass/fail
diff --git a/tests/integration/functional/qc/test_xcp.py b/tests/integration/functional/qc/test_xcp.py
@@ -72,15 +72,11 @@ def test_xcp_qc_from_bold(
     assert metrics.meanDVInit >= 0
 
     # Write TSV and verify
-    out_path = tmp_path / "xcp_qc.tsv"
+    out_path = tmp_path / "xcp_qc.parquet"
     write_xcp_qc(metrics, out_path)
     assert out_path.exists()
 
-    df = pl.read_csv(
-        out_path,
-        separator="\t",
-        schema_overrides={"sub": pl.Utf8, "ses": pl.Utf8},
-    )
+    df = pl.read_parquet(out_path)
     assert df.shape == (1, 24)
     assert df["sub"][0] == test_subject.subject_id
     assert df["meanFD"][0] == metrics.meanFD
diff --git a/tests/integration/test_all.py b/tests/integration/test_all.py
@@ -185,14 +185,14 @@ def _assert_derivatives_exist(output_dir: Path) -> None:
         )
 
     # -- QC --
-    qc_files = list(func.glob(f"{bold_stem}_space-*_*_quality.tsv"))
+    qc_files = list(func.glob(f"{bold_stem}_space-*_*_quality.parquet"))
     assert qc_files, f"No QC quality TSV files found\n--- file tree ---\n{tree}"
 
     # -- Metrics --
-    assert list(func.glob(f"{bold_stem}_space-*_*_timeseries.tsv")), (
+    assert list(func.glob(f"{bold_stem}_space-*_*_timeseries.parquet")), (
         f"No timeseries TSV files found\n--- file tree ---\n{tree}"
     )
-    assert list(func.glob(f"{bold_stem}_space-*_*_correlations.tsv")), (
+    assert list(func.glob(f"{bold_stem}_space-*_*_connectome.parquet")), (
         f"No correlation matrix TSV files found\n--- file tree ---\n{tree}"
     )
 
diff --git a/tests/unit/bids/test_exports.py b/tests/unit/bids/test_exports.py
@@ -89,8 +89,8 @@ def _make_metrics_outputs(w: Path, atlases: list[str]) -> MetricsOutputs:
         reho=_dummy(w, "reho.nii.gz"),
         reho_smooth=_dummy(w, "reho_smooth.nii.gz"),
         reho_zscored=_dummy(w, "reho_z.nii.gz"),
-        timeseries={a: _dummy(w, f"ts_{a}.tsv") for a in atlases},
-        correlation_matrix={a: _dummy(w, f"corr_{a}.tsv") for a in atlases},
+        timeseries={a: _dummy(w, f"ts_{a}.parquet") for a in atlases},
+        connectome={a: _dummy(w, f"connectome_{a}.parquet") for a in atlases},
     )
 
 
@@ -294,9 +294,9 @@ class _FakeQC:
             qc_file: dict[str, Path] = field(default_factory=dict)
 
         mni = func_bids.derive(space="MNI152NLin6Asym")
-        qc = _FakeQC(qc_file={"36-parameter": _dummy(workdir, "qc.tsv")})
+        qc = _FakeQC(qc_file={"36-parameter": _dummy(workdir, "qc.parquet")})
         export_qc(mni, qc, regressors=["36-parameter"])  # type: ignore[arg-type]
-        saved = list(pipe_ctx.output_dir.rglob("*.tsv"))
+        saved = list(pipe_ctx.output_dir.rglob("*.parquet"))
         assert len(saved) == 1
         assert "reg-36parameter" in saved[0].name
 
@@ -312,7 +312,7 @@ class _FakeQC:
 
         regs = ["36-parameter", "aCompCor"]
         mni = func_bids.derive(space="MNI152NLin6Asym")
-        qc = _FakeQC(qc_file={r: _dummy(workdir, f"qc_{r}.tsv") for r in regs})
+        qc = _FakeQC(qc_file={r: _dummy(workdir, f"qc_{r}.parquet") for r in regs})
         export_qc(mni, qc, regressors=regs)  # type: ignore[arg-type]
-        saved = list(pipe_ctx.output_dir.rglob("*.tsv"))
+        saved = list(pipe_ctx.output_dir.rglob("*.parquet"))
         assert len(saved) == 2
diff --git a/tests/unit/core/qc/test_xcp.py b/tests/unit/core/qc/test_xcp.py
@@ -110,8 +110,8 @@ def _sample_xcp_metrics() -> XCPQCMetrics:
 class TestXCPQCMetrics:
     """Tests for XCPQCMetrics NamedTuple structure."""
 
-    def test_field_names_match_tsv_columns(self) -> None:
-        """NamedTuple field names exactly match expected TSV columns."""
+    def test_field_names_match_pq_columns(self) -> None:
+        """NamedTuple field names exactly match expected Parquet columns."""
         assert list(XCPQCMetrics._fields) == EXPECTED_COLUMNS
 
     def test_field_count(self) -> None:
@@ -224,55 +224,51 @@ class TestWriteXcpQc:
 
     def test_writes_file(self, tmp_path: Path) -> None:
         """Output file is created."""
-        out = tmp_path / "qc.tsv"
+        out = tmp_path / "qc.parquet"
         result = write_xcp_qc(_sample_xcp_metrics(), out)
         assert result == out
         assert out.exists()
 
     def test_correct_headers(self, tmp_path: Path) -> None:
-        """TSV header row matches expected column names."""
-        out = tmp_path / "qc.tsv"
+        """Parquet column names match expected column names."""
+        out = tmp_path / "qc.parquet"
         write_xcp_qc(_sample_xcp_metrics(), out)
-        header = out.read_text().splitlines()[0].split("\t")
-        assert header == EXPECTED_COLUMNS
+        df = pl.read_parquet(out)
+        assert df.columns == EXPECTED_COLUMNS
 
     def test_correct_values(self, tmp_path: Path) -> None:
         """Values in TSV match the input metrics."""
         m = _sample_xcp_metrics()
-        out = tmp_path / "qc.tsv"
+        out = tmp_path / "qc.parquet"
         write_xcp_qc(m, out)
-        values = out.read_text().splitlines()[1].split("\t")
-        assert values[0] == "01"  # sub
-        assert values[3] == "1"  # run
-        assert float(values[7]) == m.meanFD
+        df = pl.read_parquet(out)
+        assert df["sub"][0] == "01"  # sub
+        assert df["run"][0] == 1  # run
+        assert df["meanFD"][0] == m.meanFD
 
     def test_round_trip_polars(self, tmp_path: Path) -> None:
         """Polars can read back the TSV and recover the values."""
         m = _sample_xcp_metrics()
-        out = tmp_path / "qc.tsv"
+        out = tmp_path / "qc.parquet"
         write_xcp_qc(m, out)
-        df = pl.read_csv(
-            out,
-            separator="\t",
-            schema_overrides={"sub": pl.Utf8, "ses": pl.Utf8},
-        )
+        df = pl.read_parquet(out)
         assert df.shape == (1, 24)
         assert df["sub"][0] == "01"
         assert df["meanFD"][0] == m.meanFD
         assert df["normCoverage"][0] == m.normCoverage
 
     def test_creates_parent_dirs(self, tmp_path: Path) -> None:
         """Parent directories are created if they don't exist."""
-        out = tmp_path / "a" / "b" / "c" / "qc.tsv"
+        out = tmp_path / "a" / "b" / "c" / "qc.parquet"
         write_xcp_qc(_sample_xcp_metrics(), out)
         assert out.exists()
 
     def test_single_data_row(self, tmp_path: Path) -> None:
-        """TSV has exactly one header row and one data row."""
-        out = tmp_path / "qc.tsv"
+        """Parquet has exactly one data row."""
+        out = tmp_path / "qc.parquet"
         write_xcp_qc(_sample_xcp_metrics(), out)
-        lines = out.read_text().strip().splitlines()
-        assert len(lines) == 2
+        df = pl.read_parquet(out)
+        assert df.shape[0] == 1
 
 
 # ===================================================================
diff --git a/tests/unit/core/test_timeseries.py b/tests/unit/core/test_timeseries.py
@@ -5,6 +5,7 @@
 from typing import TYPE_CHECKING
 
 import numpy as np
+import polars as pl
 import pytest
 
 from rbc.core.metrics.timeseries import (
@@ -247,7 +248,7 @@ def _make_nifti(self, data: np.ndarray, path: Path) -> None:
         img.to_filename(str(path))
 
     def test_round_trip(self, tmp_path: Path) -> None:
-        """Should produce TSV files that can be loaded back."""
+        """Should produce Parquet files that can be loaded back."""
         rng = np.random.default_rng(20)
         data = rng.standard_normal((4, 4, 4, 10))
         atlas = np.zeros((4, 4, 4), dtype=np.int16)
@@ -262,13 +263,13 @@ def test_round_trip(self, tmp_path: Path) -> None:
         result = compute_timeseries(in_file, atlas_file)
 
         assert result.timeseries.exists()
-        assert result.correlation_matrix.exists()
+        assert result.connectome.exists()
         assert len(result.labels) == 2
 
-        ts_loaded = np.loadtxt(result.timeseries, delimiter="\t")
-        assert ts_loaded.shape == (2, 10)
+        ts_loaded = pl.read_parquet(result.timeseries)
+        assert ts_loaded.shape == (10, 2)
 
-        corr_loaded = np.loadtxt(result.correlation_matrix, delimiter="\t")
+        corr_loaded = pl.read_parquet(result.connectome)
         assert corr_loaded.shape == (2, 2)
 
     def test_output_naming(self, tmp_path: Path) -> None:
@@ -285,8 +286,8 @@ def test_output_naming(self, tmp_path: Path) -> None:
 
         result = compute_timeseries(in_file, atlas_file)
 
-        assert result.timeseries.name == "sub-01_bold_timeseries.tsv"
-        assert result.correlation_matrix.name == "sub-01_bold_correlation_matrix.tsv"
+        assert result.timeseries.name == "sub-01_bold_timeseries.parquet"
+        assert result.connectome.name == "sub-01_bold_connectome.parquet"
 
     def test_custom_out_dir(self, tmp_path: Path) -> None:
         """Should write to a custom output directory."""
@@ -304,7 +305,7 @@ def test_custom_out_dir(self, tmp_path: Path) -> None:
         result = compute_timeseries(in_file, atlas_file, out_dir=out_dir)
 
         assert result.timeseries.parent == out_dir
-        assert result.correlation_matrix.parent == out_dir
+        assert result.connectome.parent == out_dir
 
     def test_labels_in_output(self, tmp_path: Path) -> None:
         """Output labels should match the atlas ROI labels."""
diff --git a/tests/unit/orchestration/test_qc.py b/tests/unit/orchestration/test_qc.py
@@ -24,7 +24,7 @@ def _mock_qc_outputs(
 ) -> QCOutputs:
     return QCOutputs(
         metrics={regressor: Mock(spec=XCPQCMetrics)},
-        qc_file={regressor: Path("fake_workdir") / "qc.tsv"},
+        qc_file={regressor: Path("fake_workdir") / "qc.parquet"},
         passed=passed,
     )
 

Original file line number	Diff line number	Diff line change
`@@ -114,6 +114,6 @@ def export_qc(`
`114`	`114`	`outputs.qc_file[reg],`
`115`	`115`	`suffix="quality",`
`116`	`116`	`desc="xcp",`
`117`		`- extension=".tsv",`
	`117`	`+ extension=".parquet",`
`118`	`118`	`extra={"reg": bids_safe_label(reg)},`
`119`	`119`	`)`
Original file line number	Diff line number	Diff line change
`@@ -160,7 +160,7 @@ def single_session_qc(`
`160`	`160`	`# 9. Write QC TSV`
`161`	`161`	`qc_outputs.qc_file[regressor] = write_xcp_qc(`
`162`	`162`	`qc_outputs.metrics[regressor],`
`163`		`- work_dir / f"reg-{regressor}_qc.tsv",`
	`163`	`+ work_dir / f"reg-{regressor}_qc.parquet",`
`164`	`164`	`)`
`165`	`165`
`166`	`166`	`# 10. RBC pass/fail`
Original file line number	Diff line number	Diff line change
`@@ -185,14 +185,14 @@ def _assert_derivatives_exist(output_dir: Path) -> None:`
`185`	`185`	`)`
`186`	`186`
`187`	`187`	`# -- QC --`
`188`		`- qc_files = list(func.glob(f"{bold_stem}_space-__quality.tsv"))`
	`188`	`+ qc_files = list(func.glob(f"{bold_stem}_space-__quality.parquet"))`
`189`	`189`	`assert qc_files, f"No QC quality TSV files found\n--- file tree ---\n{tree}"`
`190`	`190`
`191`	`191`	`# -- Metrics --`
`192`		`- assert list(func.glob(f"{bold_stem}_space-__timeseries.tsv")), (`
	`192`	`+ assert list(func.glob(f"{bold_stem}_space-__timeseries.parquet")), (`
`193`	`193`	`f"No timeseries TSV files found\n--- file tree ---\n{tree}"`
`194`	`194`	`)`
`195`		`- assert list(func.glob(f"{bold_stem}_space-__correlations.tsv")), (`
	`195`	`+ assert list(func.glob(f"{bold_stem}_space-__connectome.parquet")), (`
`196`	`196`	`f"No correlation matrix TSV files found\n--- file tree ---\n{tree}"`
`197`	`197`	`)`
`198`	`198`