Add metadata loading stage with TR validation (#257)

* Add metadata loading stage with TR validation

Introduce FunctionalMetadata frozen dataclass that loads, validates,
and logs TR from BIDS sidecar and NIfTI header before any processing
begins. Add --tr CLI override to functional, all, and metrics commands.

Closes #225, closes #224

* Add plausibility checks for TR and SliceTiming

Warn on implausible TR values (outside 0.1-20s range) and raise on
SliceTiming values outside [0, TR) per the BIDS spec.

* Fix full_pipeline tests for new metadata/TR signatures

* Remove compat alias

* All the renames

* Ruff format

* Duplicate import

* Remove useless test

Author: nx10

scripts/generate_bids_tools.py

@@ -255,7 +255,7 @@ def w(s: str = "") -> None:
     w()
 
     # Keys excluded from entity kwargs TypedDicts because they are handled
-    # as explicit parameters (sub/ses from PipelineContext, desc/space/atlas
+    # as explicit parameters (sub/ses from RunContext, desc/space/atlas
     # vary per-call rather than per-session).
     excluded_entity_keys = {"sub", "ses", "desc", "space", "atlas"}
 

src/rbc/cli/all.py

@@ -22,7 +22,7 @@
 )
 from rbc.cli.base import BaseArgs, _validate_atlas, _validate_positive, _validate_task
 from rbc.cli.query import iter_session_files, load_session
-from rbc.context import PipelineContext
+from rbc.context import RunContext
 from rbc.core.bids import (
     Datatype,
     Suffix,
@@ -32,6 +32,7 @@
 )
 from rbc.core.bids2table import load_table
 from rbc.core.niwrap import setup_runner
+from rbc.metadata import FunctionalMetadata
 from rbc.workflows.anatomical import single_session_preprocess as anatomical_preprocess
 from rbc.workflows.functional import single_session_preprocess as functional_preprocess
 from rbc.workflows.metrics import single_session_metrics as metrics_pipeline
@@ -53,6 +54,7 @@ class AllArgs(BaseArgs):
     atlas: Sequence[AtlasName]
     fwhm: float
     start_tr: int
+    tr: float | None
 
     @classmethod
     def validate_namespace(cls, ns: argparse.Namespace) -> AllArgs:
@@ -62,13 +64,15 @@ def validate_namespace(cls, ns: argparse.Namespace) -> AllArgs:
             _validate_atlas(atlas)
         _validate_positive(ns.fwhm, "FWHM")
         _validate_positive(ns.start_tr, "Start TR")
+        _validate_positive(ns.tr, "TR")
         return cls(
             **BaseArgs.validate_namespace(ns).__dict__,
             regressor=ns.regressor,
             task=ns.task,
             atlas=ns.atlas,
             fwhm=ns.fwhm,
             start_tr=ns.start_tr,
+            tr=ns.tr,
         )
 
 
@@ -98,7 +102,7 @@ def main(args: AllArgs) -> int:  # noqa: C901
     for _, sub_ses_group in tqdm(
         df.group_by(_SUB_SES_QUERY, maintain_order=True), disable=not ctx.verbose
     ):
-        pipe_ctx = PipelineContext(
+        pipe_ctx = RunContext(
             sub=sub_ses_group["sub"][0],
             ses=sub_ses_group["ses"][0] or None,
             output_dir=args.output_dir,
@@ -151,6 +155,8 @@ def main(args: AllArgs) -> int:  # noqa: C901
             ents = extract_entities(row, ["task", "run", "acq", "rec", "dir", "echo"])
             ctx.logger.info(f"Functional: {bold_fpath}")
 
+            func_metadata = FunctionalMetadata.load(bold_fpath, tr_override=args.tr)
+
             func_outputs = functional_preprocess(
                 in_bold=bold_fpath,
                 t1w_brain=anat_outputs.brain,
@@ -159,6 +165,7 @@ def main(args: AllArgs) -> int:  # noqa: C901
                 csf_mask=anat_outputs.csf_mask,
                 wm_mask=anat_outputs.wm_mask,
                 anat_to_template=anat_outputs.inverse_xfm,
+                metadata=func_metadata,
                 start_tr=args.start_tr,
                 regressor_set=args.regressor,
             )
@@ -236,6 +243,7 @@ def main(args: AllArgs) -> int:  # noqa: C901
                     regressed_bold=func_outputs.regressed_bold[regressor],
                     cleaned_bold=func_outputs.cleaned_bold[regressor],
                     template_brain_mask=func_outputs.template_brain_mask,
+                    tr=func_metadata.tr,
                     atlas=args.atlas,
                     fwhm=args.fwhm,
                 )
@@ -358,5 +366,11 @@ def register_command(
         default=2,
         help="Number of initial TRs to discard.",
     )
+    parser.add_argument(
+        "--tr",
+        type=float,
+        default=None,
+        help="Repetition time in seconds. Overrides BIDS sidecar and NIfTI header.",
+    )
 
     parser.set_defaults(func=lambda args: main(AllArgs.validate_namespace(args)))

src/rbc/cli/anatomical.py

@@ -18,7 +18,7 @@
 
 from rbc.cli import _ANAT_GROUP_ENTITIES, _DEFAULT_ENV_VARS, _SUB_SES_QUERY
 from rbc.cli.base import BaseArgs
-from rbc.context import PipelineContext
+from rbc.context import RunContext
 from rbc.core.bids import Datatype, Suffix, TemplateSpace, extract_entities
 from rbc.core.bids2table import load_table
 from rbc.core.niwrap import setup_runner
@@ -60,7 +60,7 @@ def main(args: AnatomicalArgs) -> int:
     for _, sub_ses_group in tqdm(
         df.group_by(_SUB_SES_QUERY, maintain_order=True), disable=not ctx.verbose
     ):
-        pipe_ctx = PipelineContext(
+        pipe_ctx = RunContext(
             sub=sub_ses_group["sub"][0],
             ses=sub_ses_group["ses"][0] or None,
             output_dir=args.output_dir,

src/rbc/cli/functional.py

@@ -17,9 +17,9 @@
 from tqdm import tqdm
 
 from rbc.cli import _DEFAULT_ENV_VARS, _FUNC_GROUP_ENTITIES, _SUB_SES_QUERY
-from rbc.cli.base import BaseArgs, _validate_task
+from rbc.cli.base import BaseArgs, _validate_positive, _validate_task
 from rbc.cli.query import iter_session_files, load_session
-from rbc.context import PipelineContext
+from rbc.context import RunContext
 from rbc.core.bids import (
     Datatype,
     Suffix,
@@ -29,6 +29,7 @@
 )
 from rbc.core.bids2table import load_table
 from rbc.core.niwrap import setup_runner
+from rbc.metadata import FunctionalMetadata
 from rbc.workflows.functional import single_session_preprocess
 
 if TYPE_CHECKING:
@@ -42,15 +43,18 @@ class FunctionalArgs(BaseArgs):
 
     regressor: Sequence[Literal["36-parameter", "aCompCor"]]
     task: str | None
+    tr: float | None
 
     @classmethod
     def validate_namespace(cls, ns: argparse.Namespace) -> FunctionalArgs:
         """Validation of functional workflow specific arguments to NamedTuple."""
         _validate_task(ns.task)
+        _validate_positive(ns.tr, "TR")
         return cls(
             **BaseArgs.validate_namespace(ns).__dict__,
             regressor=ns.regressor,  # Validated by argparse choices
             task=ns.task,
+            tr=ns.tr,
         )
 
 
@@ -77,7 +81,7 @@ def main(args: FunctionalArgs) -> int:
     for _, sub_ses_group in tqdm(
         df.group_by(_SUB_SES_QUERY, maintain_order=True), disable=not ctx.verbose
     ):
-        pipe_ctx = PipelineContext(
+        pipe_ctx = RunContext(
             sub=sub_ses_group["sub"][0],
             ses=sub_ses_group["ses"][0] or None,
             output_dir=args.output_dir,
@@ -96,6 +100,8 @@ def main(args: FunctionalArgs) -> int:
 
             anat_q = pipe_ctx.bids(datatype=Datatype.ANAT)
 
+            func_metadata = FunctionalMetadata.load(bold_fpath, tr_override=args.tr)
+
             outputs = single_session_preprocess(
                 in_bold=bold_fpath,
                 t1w_brain=anat_q.expect(anat_df, suffix=Suffix.T1W, desc="brain"),
@@ -112,6 +118,7 @@ def main(args: FunctionalArgs) -> int:
                         "mode": "image",
                     },
                 ),
+                metadata=func_metadata,
                 regressor_set=args.regressor,
             )
 
@@ -205,5 +212,11 @@ def register_command(
         default=None,
         help="Task label to filter BOLD runs (without 'task-' prefix).",
     )
+    parser.add_argument(
+        "--tr",
+        type=float,
+        default=None,
+        help="Repetition time in seconds. Overrides BIDS sidecar and NIfTI header.",
+    )
 
     parser.set_defaults(func=lambda args: main(FunctionalArgs.validate_namespace(args)))

src/rbc/cli/longitudinal.py

@@ -16,7 +16,7 @@
 from rbc.cli import _DEFAULT_ENV_VARS, _FUNC_GROUP_ENTITIES, _SUB_SES_QUERY
 from rbc.cli.base import BaseArgs
 from rbc.cli.query import iter_session_files, load_session
-from rbc.context import PipelineContext
+from rbc.context import RunContext
 from rbc.core.bids import Datatype, Extension, Suffix, extract_entities
 from rbc.core.bids2table import load_table
 from rbc.core.niwrap import setup_runner
@@ -52,7 +52,7 @@ def _require_file(path: Path | None, field: str) -> Path:
 
 
 def _process_anat(
-    pipe_ctx: PipelineContext, anat_df: pl.DataFrame, tpl_df: pl.DataFrame
+    pipe_ctx: RunContext, anat_df: pl.DataFrame, tpl_df: pl.DataFrame
 ) -> None:
     """Handle anatomical longitudinal processing."""
     anat_df = anat_df.filter(pl.col("space").is_null())
@@ -102,7 +102,7 @@ def _process_anat(
 
 
 def _process_func(
-    pipe_ctx: PipelineContext, func_df: pl.DataFrame, tpl_df: pl.DataFrame
+    pipe_ctx: RunContext, func_df: pl.DataFrame, tpl_df: pl.DataFrame
 ) -> None:
     """Handle functional longitudinal processing."""
     row = func_df.filter(suffix=Suffix.BOLD).row(0, named=True)
@@ -174,7 +174,7 @@ def main(args: LongitudinalArgs) -> int:
     for _, sub_ses_group in tqdm(
         group_df.group_by(_SUB_SES_QUERY, maintain_order=True), disable=not ctx.verbose
     ):
-        pipe_ctx = PipelineContext(
+        pipe_ctx = RunContext(
             sub=sub_ses_group["sub"][0],
             ses=sub_ses_group["ses"][0],
             output_dir=args.output_dir,

src/rbc/cli/metrics.py

@@ -6,15 +6,17 @@
 
 from __future__ import annotations
 
+import logging
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Literal
 
+import nibabel as nib
 import polars as pl
 from tqdm import tqdm
 
 from rbc.cli import _DEFAULT_ENV_VARS, _FUNC_GROUP_ENTITIES, _SUB_SES_QUERY
 from rbc.cli.base import BaseArgs, _validate_atlas, _validate_positive, _validate_task
-from rbc.context import PipelineContext
+from rbc.context import RunContext
 from rbc.core.bids import Datatype, Suffix, TemplateSpace, extract_entities
 from rbc.core.bids2table import load_table
 from rbc.core.niwrap import setup_runner
@@ -23,6 +25,7 @@
 if TYPE_CHECKING:
     import argparse
     from collections.abc import Sequence
+    from pathlib import Path
 
     from rbc_resources import AtlasName
 
@@ -35,6 +38,7 @@ class MetricsArgs(BaseArgs):
     fwhm: float
     task: str | None
     regressor: Sequence[Literal["36-parameter", "aCompCor"]]
+    tr: float | None
 
     @classmethod
     def validate_namespace(cls, ns: argparse.Namespace) -> MetricsArgs:
@@ -43,21 +47,43 @@ def validate_namespace(cls, ns: argparse.Namespace) -> MetricsArgs:
             _validate_atlas(atlas)
         _validate_task(ns.task)
         _validate_positive(ns.fwhm, "FWHM")
+        _validate_positive(ns.tr, "TR")
         return cls(
             **BaseArgs.validate_namespace(ns).__dict__,
             atlas=ns.atlas,
             fwhm=ns.fwhm,
             task=ns.task,
             regressor=ns.regressor,
+            tr=ns.tr,
         )
 
 
+_logger = logging.getLogger(__name__)
+
+
+def _read_header_tr(nifti_path: Path) -> float:
+    """Read TR from a NIfTI header, raising on missing/zero values."""
+    hdr = nib.nifti1.load(nifti_path).header
+    tr = float(hdr["pixdim"][4])  # type: ignore[index]
+    if tr <= 0:
+        msg = (
+            f"NIfTI header TR is {tr} for {nifti_path}. Pass --tr to specify manually."
+        )
+        raise ValueError(msg)
+    _logger.info("TR: %.4f s (from NIfTI header)", tr)
+    return tr
+
+
 def main(args: MetricsArgs) -> int:
     """Main entrypoint of metrics workflow."""
     ctx = setup_runner(runner=args.runner, verbose=args.verbose, tmp_dir=args.tmp_dir)
     ctx.runner.environ = _DEFAULT_ENV_VARS
 
     ctx.logger.info("Preparing to run RBC metrics workflow")
+    if args.tr is not None:
+        ctx.logger.info("Using CLI-provided TR: %.4f s", args.tr)
+    else:
+        ctx.logger.info("TR will be read from NIfTI headers")
     df = load_table(
         dataset_dir=args.output_dir,
         index_fpath=None,
@@ -82,7 +108,7 @@ def main(args: MetricsArgs) -> int:
     for _, group in tqdm(df.group_by(_SUB_SES_QUERY), disable=not ctx.verbose):
         sub: str = group["sub"][0]
         ses: str | None = group["ses"][0] or None
-        pipe_ctx = PipelineContext(sub=sub, ses=ses, output_dir=args.output_dir)
+        pipe_ctx = RunContext(sub=sub, ses=ses, output_dir=args.output_dir)
 
         deriv_df = load_table(
             dataset_dir=args.output_dir, index_fpath=None, max_workers=0, verbose=False
@@ -115,10 +141,13 @@ def main(args: MetricsArgs) -> int:
                     extra={"reg": regressor},
                 )
 
+                tr = args.tr if args.tr is not None else _read_header_tr(regressed_bold)
+
                 outputs = single_session_metrics(
                     regressed_bold=regressed_bold,
                     cleaned_bold=cleaned_bold,
                     template_brain_mask=template_brain_mask,
+                    tr=tr,
                     atlas=args.atlas,
                     fwhm=args.fwhm,
                 )
@@ -199,5 +228,11 @@ def register_command(
             "functional preprocessing."
         ),
     )
+    parser.add_argument(
+        "--tr",
+        type=float,
+        default=None,
+        help="Repetition time in seconds. Overrides NIfTI header value for ALFF.",
+    )
 
     parser.set_defaults(func=lambda args: main(MetricsArgs.validate_namespace(args)))

src/rbc/cli/qc.py

@@ -15,7 +15,7 @@
 
 from rbc.cli import _DEFAULT_ENV_VARS, _FUNC_GROUP_ENTITIES, _SUB_SES_QUERY
 from rbc.cli.base import BaseArgs, _validate_positive, _validate_task
-from rbc.context import PipelineContext
+from rbc.context import RunContext
 from rbc.core.bids import Datatype, Suffix, TemplateSpace, extract_entities
 from rbc.core.bids2table import load_table
 from rbc.core.niwrap import setup_runner
@@ -78,7 +78,7 @@ def main(args: QCArgs) -> int:
     for _, group in tqdm(df.group_by(_SUB_SES_QUERY), disable=not ctx.verbose):
         sub: str = group["sub"][0]
         ses: str | None = group["ses"][0] or None
-        pipe_ctx = PipelineContext(sub=sub, ses=ses, output_dir=args.output_dir)
+        pipe_ctx = RunContext(sub=sub, ses=ses, output_dir=args.output_dir)
 
         deriv_df = load_table(
             dataset_dir=args.output_dir, index_fpath=None, max_workers=0, verbose=False

src/rbc/context.py

@@ -3,6 +3,7 @@
 Holds subject identity and output directory, providing a :meth:`bids` factory
 that returns a :class:`~rbc.core.bids.Bids` builder for composing exports
 and queries.
+
 """
 
 from __future__ import annotations
@@ -20,8 +21,8 @@
 _RBC_VERSION = version("rbc")
 
 
-@dataclass
-class PipelineContext:
+@dataclass(frozen=True)
+class RunContext:
     """Minimal context for a single pipeline run.
 
     Attributes: