Metrics, QC & all pipeline in longitudinal space (Stage 6 of #301) (#317)

* Add longitudinal metrics orchestration (port from #297)

Resolve per-regressor regressed/cleaned BOLD and brain mask from
space-longitudinal derivatives, read TR from NIfTI header, call
single_session_metrics unchanged. Reuse cross-sectional export_metrics
for BIDS naming.

Co-authored-by: Janhavi Pillai <janhavipillai3@gmail.com>

* Add longitudinal QC, all pipeline, and full_pipeline tests

QC: Dice/Jaccard between anat and BOLD brain masks in longitudinal
space, pass threshold Dice >= 0.85. Viz tracked in #303/#304.

All pipeline: template -> anat -> func -> metrics -> qc. Template
writes to disk cross-session; per-session stages hand off func outputs
in-memory. process_anat/process_func now return workflow outputs.

CLI: Add --regressor and --task to `rbc longitudinal all`.

Tests: Tier-4 full_pipeline tests for metrics, QC, and all-vs-sequential
equivalence under tests/full_pipeline/longitudinal/.

* Fix anat groupby dropping mask rows in longitudinal resolve

T1w suffix filter in run() excluded masks from the DataFrame passed to
process_anat, so resolve_longitudinal_anat couldn't find them. Drop the
filter. Pre-existing Stage 4 bug exposed by new tests.

* Add --regressor to longitudinal metrics CLI

Replace DataFrame-based discovery (queried a non-existent 'reg' column)
with explicit --regressor, matching cross-sectional.

* Fix longitudinal all test fixture missing --session-label

Fixture ran cross-sectional func only for ses-test but invoked
`rbc longitudinal all` without --session-label, so it iterated ses-retest
(no func derivatives) and crashed on an empty BOLD DataFrame.

* Apply only participant filter to template discovery in all pipeline

Template needs all sessions to find multi-session subjects. Session/task
filters apply only to per-session stages, not discover_template_inputs.

* Clean up tech debt: TR validation, FS auth, QC load_table

TR: Promote resolve_tr/warn_implausible_tr to public API. Replace
_read_header_tr with _read_derivative_tr that pipes through resolve_tr
for validation and plausibility warnings. Add --tr override to
longitudinal metrics CLI.

FS auth: Rename _setup_freesurfer_auth -> setup_freesurfer_auth, export
from template.py, import at module level in all.py.

QC: Replace manual concat of session.anat + session.func + tpl_df with
load_table, matching cross-sectional qc.py.

---------
Co-authored-by: Janhavi Pillai <janhavi.pillai@gmail.com>

Author: nx10

src/rbc/bids/longitudinal/metrics.py

@@ -0,0 +1,64 @@
+"""BIDS resolve and export for the longitudinal metrics workflow.
+
+Reuses the cross-sectional :func:`~rbc.bids.metrics.export_metrics` for
+export since the output structure (ALFF, fALFF, ReHo, timeseries,
+correlations) is identical -- only the space changes to ``longitudinal``.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, TypedDict
+
+from rbc.bids import Suffix, bids_safe_label
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    import polars as pl
+
+    from rbc.bids import Bids
+
+
+class MetricsLongInputs(TypedDict):
+    """Resolved functional inputs for the longitudinal metrics workflow."""
+
+    template_brain_mask: Path
+    regressed_bold: Path
+    cleaned_bold: Path
+
+
+def resolve_longitudinal_metrics(
+    func_long_q: Bids,
+    func_long_df: pl.DataFrame,
+    *,
+    regressor: str,
+) -> MetricsLongInputs:
+    """Resolve longitudinal-space functional derivatives for metrics.
+
+    Args:
+        func_long_q: Bids builder configured for ``space=longitudinal``
+            func queries.
+        func_long_df: DataFrame of longitudinal-space derivative outputs.
+        regressor: Single regressor name (e.g. ``"36-parameter"``).
+
+    Returns:
+        Dict with keys: ``template_brain_mask``, ``regressed_bold``,
+        ``cleaned_bold``.
+    """
+    return {
+        "template_brain_mask": func_long_q.expect(
+            func_long_df, suffix=Suffix.MASK, desc="brain"
+        ),
+        "regressed_bold": func_long_q.expect(
+            func_long_df,
+            suffix=Suffix.BOLD,
+            desc="regressed",
+            extra={"reg": bids_safe_label(regressor)},
+        ),
+        "cleaned_bold": func_long_q.expect(
+            func_long_df,
+            suffix=Suffix.BOLD,
+            desc="preproc",
+            extra={"reg": bids_safe_label(regressor)},
+        ),
+    }

src/rbc/bids/longitudinal/qc.py

@@ -0,0 +1,149 @@
+"""BIDS resolve and export for the longitudinal QC workflow.
+
+Longitudinal QC is minimal: Dice/Jaccard overlap between the anatomical
+brain mask and BOLD brain mask, both in longitudinal template space.
+"""
+
+from __future__ import annotations
+
+import csv
+from typing import TYPE_CHECKING, NamedTuple, TypedDict
+
+from rbc.bids import Suffix
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    import polars as pl
+
+    from rbc.bids import Bids
+
+
+class QCLongInputs(TypedDict):
+    """Resolved inputs for the longitudinal registration QC."""
+
+    anat_brain_mask: Path
+    bold_mask: Path
+
+
+class LongitudinalQCOutputs(NamedTuple):
+    """QC outputs from the longitudinal registration overlap check.
+
+    Attributes:
+        dice: Dice coefficient between anat and BOLD masks.
+        jaccard: Jaccard index between anat and BOLD masks.
+        coverage: Coverage of the smaller mask by the overlap.
+        cross_corr: Pearson correlation between flattened masks.
+        passed: Whether the run passes the Dice threshold.
+        qc_file: Path to the written single-row QC TSV.
+    """
+
+    dice: float
+    jaccard: float
+    coverage: float
+    cross_corr: float
+    passed: bool
+    qc_file: Path
+
+
+def resolve_longitudinal_qc(
+    anat_long_q: Bids,
+    func_long_q: Bids,
+    anat_long_df: pl.DataFrame,
+    func_long_df: pl.DataFrame,
+) -> QCLongInputs:
+    """Resolve longitudinal-space masks for registration QC.
+
+    Args:
+        anat_long_q: Bids builder for ``space=longitudinal`` anat queries.
+        func_long_q: Bids builder for ``space=longitudinal`` func queries.
+        anat_long_df: DataFrame of longitudinal anatomical derivatives.
+        func_long_df: DataFrame of longitudinal functional derivatives.
+
+    Returns:
+        Dict with ``anat_brain_mask`` and ``bold_mask`` paths.
+    """
+    return {
+        "anat_brain_mask": anat_long_q.expect(
+            anat_long_df, suffix=Suffix.MASK, desc="T1w"
+        ),
+        "bold_mask": func_long_q.expect(func_long_df, suffix=Suffix.MASK, desc="brain"),
+    }
+
+
+def export_longitudinal_qc(
+    func_long: Bids,
+    outputs: LongitudinalQCOutputs,
+) -> None:
+    """Export longitudinal QC results as a single-row TSV.
+
+    Args:
+        func_long: Bids builder with ``space="longitudinal"`` for func.
+        outputs: QC overlap metrics.
+    """
+    func_long.save(
+        outputs.qc_file,
+        suffix="quality",
+        desc="registration",
+        extension=".tsv",
+    )
+
+
+def write_longitudinal_qc_tsv(
+    out_path: Path,
+    *,
+    sub: str,
+    ses: str,
+    task: str,
+    run: int | str,
+    dice: float,
+    jaccard: float,
+    coverage: float,
+    cross_corr: float,
+    passed: bool,
+) -> Path:
+    """Write a single-row longitudinal QC TSV.
+
+    Args:
+        out_path: Destination file path.
+        sub: Subject ID.
+        ses: Session label.
+        task: Task label.
+        run: Run number.
+        dice: Dice coefficient.
+        jaccard: Jaccard index.
+        coverage: Coverage metric.
+        cross_corr: Cross-correlation.
+        passed: Pass/fail flag.
+
+    Returns:
+        The written file path.
+    """
+    fieldnames = [
+        "sub",
+        "ses",
+        "task",
+        "run",
+        "dice",
+        "jaccard",
+        "coverage",
+        "cross_corr",
+        "passed",
+    ]
+    row = {
+        "sub": sub,
+        "ses": ses,
+        "task": task,
+        "run": run,
+        "dice": f"{dice:.6f}",
+        "jaccard": f"{jaccard:.6f}",
+        "coverage": f"{coverage:.6f}",
+        "cross_corr": f"{cross_corr:.6f}",
+        "passed": str(passed),
+    }
+    out_path.parent.mkdir(parents=True, exist_ok=True)
+    with out_path.open("w", newline="") as fh:
+        writer = csv.DictWriter(fh, fieldnames=fieldnames, delimiter="\t")
+        writer.writeheader()
+        writer.writerow(row)
+    return out_path

src/rbc/cli/longitudinal/all.py

@@ -1,11 +1,16 @@
-"""``rbc longitudinal all`` subcommand (placeholder for Stage 6)."""
+"""``rbc longitudinal all`` subcommand."""
 
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Literal
 
-from rbc.cli.base import _or_default, _validate_nifti_path, _validate_positive
+from rbc.cli.base import (
+    _or_default,
+    _validate_nifti_path,
+    _validate_positive,
+    _validate_task,
+)
 from rbc.cli.longitudinal._base import LongitudinalBaseArgs, add_fs_license_argument
 from rbc.cli.metrics import _resolve_atlas_args
 from rbc.orchestration import Filters, RunnerConfig
@@ -25,18 +30,23 @@ class AllLongArgs(LongitudinalBaseArgs):
     registration_template: Path
     atlas_files: dict[str, Path]
     fwhm: float
+    regressor: Sequence[Literal["36-parameter", "aCompCor"]]
+    task: str | None
 
     @classmethod
     def validate_namespace(cls, ns: argparse.Namespace) -> AllLongArgs:
         """Validate namespace for the full longitudinal pipeline subcommand."""
         _validate_positive(ns.fwhm, "FWHM")
+        _validate_task(ns.task)
         return cls(
             **LongitudinalBaseArgs.validate_namespace(ns).__dict__,
             registration_template=_or_default(
                 ns.anat_template, REGISTRATION_TEMPLATES.brain_1mm
             ),
             atlas_files=_resolve_atlas_args(ns.atlas),
             fwhm=ns.fwhm,
+            regressor=ns.regressor,
+            task=ns.task,
         )
 
 
@@ -48,7 +58,9 @@ def main(args: AllLongArgs) -> int:
         filters=Filters(
             participant_label=args.participant_label,
             session_label=args.session_label,
+            task=args.task,
         ),
+        regressors=args.regressor,
         fs_license=args.fs_license,
         atlas_files=args.atlas_files,
         fwhm=args.fwhm,
@@ -66,21 +78,33 @@ def register_command(
     subparsers: argparse._SubParsersAction,
     parents: Sequence[argparse.ArgumentParser],
 ) -> None:
-    """Register ``rbc longitudinal all`` (Stage 6 placeholder)."""
+    """Register ``rbc longitudinal all``."""
     parser = subparsers.add_parser(
         "all",
         parents=parents,
         description=(
-            "Run the full longitudinal pipeline (template → anat → func → "
-            "metrics → qc). Placeholder wired up by Stage 3; full "
-            "implementation ships in Stage 6."
+            "Run the full longitudinal pipeline (template -> anat -> func -> "
+            "metrics -> qc). Passes functional outputs in-memory to metrics "
+            "and QC stages."
         ),
-        help="Full longitudinal pipeline (Stage 6)",
+        help="Full longitudinal pipeline",
         usage=(
             "rbc longitudinal all INPUT_DIR [INPUT_DIR ...] -o OUTPUT_DIR [options]"
         ),
     )
     add_fs_license_argument(parser)
+    parser.add_argument(
+        "--regressor",
+        nargs="+",
+        choices=["36-parameter", "aCompCor"],
+        default=["36-parameter"],
+        help="Space-delimited nuisance regression method(s) to apply.",
+    )
+    parser.add_argument(
+        "--task",
+        default=None,
+        help="Task label to filter BOLD runs (without 'task-' prefix).",
+    )
     parser.add_argument(
         "--atlas",
         nargs="+",

src/rbc/cli/longitudinal/metrics.py

@@ -1,9 +1,9 @@
-"""``rbc longitudinal metrics`` subcommand (placeholder for Stage 6)."""
+"""``rbc longitudinal metrics`` subcommand."""
 
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Literal
 
 from rbc.cli.base import _validate_positive, _validate_task
 from rbc.cli.longitudinal._base import LongitudinalBaseArgs, add_fs_license_argument
@@ -24,18 +24,23 @@ class MetricsLongArgs(LongitudinalBaseArgs):
 
     atlas_files: dict[str, Path]
     fwhm: float
+    tr: float | None
     task: str | None
+    regressor: Sequence[Literal["36-parameter", "aCompCor"]]
 
     @classmethod
     def validate_namespace(cls, ns: argparse.Namespace) -> MetricsLongArgs:
         """Validate namespace for the longitudinal metrics subcommand."""
         _validate_task(ns.task)
         _validate_positive(ns.fwhm, "FWHM")
+        _validate_positive(ns.tr, "TR")
         return cls(
             **LongitudinalBaseArgs.validate_namespace(ns).__dict__,
             atlas_files=_resolve_atlas_args(ns.atlas),
             fwhm=ns.fwhm,
+            tr=ns.tr,
             task=ns.task,
+            regressor=ns.regressor,
         )
 
 
@@ -49,8 +54,10 @@ def main(args: MetricsLongArgs) -> int:
             session_label=args.session_label,
             task=args.task,
         ),
+        regressors=args.regressor,
         atlas_files=args.atlas_files,
         fwhm=args.fwhm,
+        tr=args.tr,
         runner_config=RunnerConfig(
             runner=args.runner,
             verbose=bool(args.verbose),
@@ -65,20 +72,27 @@ def register_command(
     subparsers: argparse._SubParsersAction,
     parents: Sequence[argparse.ArgumentParser],
 ) -> None:
-    """Register ``rbc longitudinal metrics`` (Stage 6 placeholder)."""
+    """Register ``rbc longitudinal metrics``."""
     parser = subparsers.add_parser(
         "metrics",
         parents=parents,
         description=(
-            "Compute resting-state metrics in longitudinal space. Placeholder "
-            "wired up by Stage 3; full implementation ships in Stage 6."
+            "Compute resting-state metrics (ALFF, fALFF, ReHo, atlas "
+            "timeseries) on longitudinal-space functional derivatives."
         ),
-        help="Longitudinal metrics stage (Stage 6)",
+        help="Compute resting-state metrics in longitudinal space",
         usage=(
             "rbc longitudinal metrics INPUT_DIR [INPUT_DIR ...] -o OUTPUT_DIR [options]"
         ),
     )
     add_fs_license_argument(parser)
+    parser.add_argument(
+        "--regressor",
+        nargs="+",
+        choices=["36-parameter", "aCompCor"],
+        default=["36-parameter"],
+        help=("Space-delimited nuisance regression method(s) to compute metrics for."),
+    )
     parser.add_argument(
         "--atlas",
         nargs="+",
@@ -96,6 +110,12 @@ def register_command(
         default=6.0,
         help="Smoothing kernel FWHM in mm.",
     )
+    parser.add_argument(
+        "--tr",
+        type=float,
+        default=None,
+        help="Repetition time in seconds. Overrides NIfTI header.",
+    )
     parser.add_argument(
         "--task",
         default=None,