Move runner setup and logging into orchestration layer

Orchestration run() functions now handle runner setup (init_runner)
and workflow start/complete logging. CLI modules are reduced to pure
arg parsing: construct Filters + RunnerConfig, call run(), return 0.

Introduces RunnerConfig dataclass and init_runner() in orchestration.
Moves _DEFAULT_ENV_VARS from cli/__init__.py to orchestration/.

Author: nx10

src/rbc/cli/__init__.py

@@ -7,10 +7,3 @@
 """
 
 from __future__ import annotations
-
-from rbc.core import CPAC_ANTS_SEED
-
-_DEFAULT_ENV_VARS = {
-    "ITK_GLOBAL_DEFAULT_NUMBER_OF_THREADS": "1",
-    "ANTS_RANDOM_SEED": CPAC_ANTS_SEED,
-}

src/rbc/cli/all.py

@@ -10,10 +10,8 @@
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Literal
 
-from rbc.cli import _DEFAULT_ENV_VARS
 from rbc.cli.base import BaseArgs, _validate_atlas, _validate_positive, _validate_task
-from rbc.core.niwrap import setup_runner
-from rbc.orchestration import Filters
+from rbc.orchestration import Filters, RunnerConfig
 from rbc.orchestration.all import run
 
 if TYPE_CHECKING:
@@ -56,10 +54,6 @@ def validate_namespace(cls, ns: argparse.Namespace) -> AllArgs:
 
 def main(args: AllArgs) -> int:
     """Main entrypoint of combined pipeline."""
-    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 full pipeline")
-
     run(
         input_dir=args.input_dir,
         output_dir=args.output_dir,
@@ -73,10 +67,12 @@ def main(args: AllArgs) -> int:
         fwhm=args.fwhm,
         start_tr=args.start_tr,
         tr=args.tr,
-        verbose=ctx.verbose,
+        runner_config=RunnerConfig(
+            runner=args.runner,
+            verbose=bool(args.verbose),
+            tmp_dir=args.tmp_dir,
+        ),
     )
-
-    ctx.logger.info("RBC full pipeline complete")
     return 0
 
 

src/rbc/cli/anatomical.py

@@ -5,10 +5,8 @@
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
-from rbc.cli import _DEFAULT_ENV_VARS
 from rbc.cli.base import BaseArgs
-from rbc.core.niwrap import setup_runner
-from rbc.orchestration import Filters
+from rbc.orchestration import Filters, RunnerConfig
 from rbc.orchestration.anatomical import run
 
 if TYPE_CHECKING:
@@ -28,21 +26,19 @@ def validate_namespace(cls, ns: argparse.Namespace) -> AnatomicalArgs:
 
 def main(args: AnatomicalArgs) -> int:
     """Main entrypoint of anatomical 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 anatomical workflow")
-
     run(
         input_dir=args.input_dir,
         output_dir=args.output_dir,
         filters=Filters(
             participant_label=args.participant_label,
             session_label=args.session_label,
         ),
-        verbose=ctx.verbose,
+        runner_config=RunnerConfig(
+            runner=args.runner,
+            verbose=bool(args.verbose),
+            tmp_dir=args.tmp_dir,
+        ),
     )
-
-    ctx.logger.info("RBC anatomical workflow complete")
     return 0
 
 

src/rbc/cli/functional.py

@@ -12,10 +12,8 @@
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Literal
 
-from rbc.cli import _DEFAULT_ENV_VARS
 from rbc.cli.base import BaseArgs, _validate_positive, _validate_task
-from rbc.core.niwrap import setup_runner
-from rbc.orchestration import Filters
+from rbc.orchestration import Filters, RunnerConfig
 from rbc.orchestration.functional import run
 
 if TYPE_CHECKING:
@@ -46,10 +44,6 @@ def validate_namespace(cls, ns: argparse.Namespace) -> FunctionalArgs:
 
 def main(args: FunctionalArgs) -> int:
     """Main entrypoint of functional 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 functional workflow")
-
     run(
         input_dir=args.input_dir,
         output_dir=args.output_dir,
@@ -60,10 +54,12 @@ def main(args: FunctionalArgs) -> int:
         ),
         regressors=args.regressor,
         tr=args.tr,
-        verbose=ctx.verbose,
+        runner_config=RunnerConfig(
+            runner=args.runner,
+            verbose=bool(args.verbose),
+            tmp_dir=args.tmp_dir,
+        ),
     )
-
-    ctx.logger.info("RBC functional workflow complete")
     return 0
 
 

src/rbc/cli/longitudinal.py

@@ -5,10 +5,8 @@
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
-from rbc.cli import _DEFAULT_ENV_VARS
 from rbc.cli.base import BaseArgs
-from rbc.core.niwrap import setup_runner
-from rbc.orchestration import Filters
+from rbc.orchestration import Filters, RunnerConfig
 from rbc.orchestration.longitudinal import run
 
 if TYPE_CHECKING:
@@ -39,10 +37,6 @@ def validate_namespace(cls, ns: argparse.Namespace) -> LongitudinalArgs:
 
 def main(args: LongitudinalArgs) -> int:
     """Main entrypoint of longitudinal 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 longitudinal workflow")
-
     run(
         input_dir=args.input_dir,
         output_dir=args.output_dir,
@@ -52,10 +46,12 @@ def main(args: LongitudinalArgs) -> int:
         ),
         anatomical=args.anatomical,
         functional=args.functional,
-        verbose=ctx.verbose,
+        runner_config=RunnerConfig(
+            runner=args.runner,
+            verbose=bool(args.verbose),
+            tmp_dir=args.tmp_dir,
+        ),
     )
-
-    ctx.logger.info("RBC longitudinal workflow complete")
     return 0
 
 

src/rbc/cli/metrics.py

@@ -9,10 +9,8 @@
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Literal
 
-from rbc.cli import _DEFAULT_ENV_VARS
 from rbc.cli.base import BaseArgs, _validate_atlas, _validate_positive, _validate_task
-from rbc.core.niwrap import setup_runner
-from rbc.orchestration import Filters
+from rbc.orchestration import Filters, RunnerConfig
 from rbc.orchestration.metrics import run
 
 if TYPE_CHECKING:
@@ -52,10 +50,6 @@ def validate_namespace(cls, ns: argparse.Namespace) -> MetricsArgs:
 
 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")
-
     run(
         output_dir=args.output_dir,
         filters=Filters(
@@ -67,10 +61,12 @@ def main(args: MetricsArgs) -> int:
         atlases=args.atlas,
         fwhm=args.fwhm,
         tr=args.tr,
-        verbose=ctx.verbose,
+        runner_config=RunnerConfig(
+            runner=args.runner,
+            verbose=bool(args.verbose),
+            tmp_dir=args.tmp_dir,
+        ),
     )
-
-    ctx.logger.info("RBC metrics workflow complete")
     return 0
 
 

src/rbc/cli/qc.py

@@ -10,10 +10,8 @@
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Literal
 
-from rbc.cli import _DEFAULT_ENV_VARS
 from rbc.cli.base import BaseArgs, _validate_positive, _validate_task
-from rbc.core.niwrap import setup_runner
-from rbc.orchestration import Filters
+from rbc.orchestration import Filters, RunnerConfig
 from rbc.orchestration.qc import run
 
 if TYPE_CHECKING:
@@ -44,10 +42,6 @@ def validate_namespace(cls, ns: argparse.Namespace) -> QCArgs:
 
 def main(args: QCArgs) -> int:
     """Main entrypoint of QC 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 QC workflow")
-
     run(
         output_dir=args.output_dir,
         filters=Filters(
@@ -57,10 +51,12 @@ def main(args: QCArgs) -> int:
         ),
         regressors=args.regressor,
         start_tr=args.start_tr,
-        verbose=ctx.verbose,
+        runner_config=RunnerConfig(
+            runner=args.runner,
+            verbose=bool(args.verbose),
+            tmp_dir=args.tmp_dir,
+        ),
     )
-
-    ctx.logger.info("RBC QC workflow complete")
     return 0
 
 

src/rbc/orchestration/__init__.py

@@ -1,17 +1,28 @@
 """Pipeline orchestration layer.
 
-Provides ``run()`` entry points for each workflow that handle BIDS table
-loading, filtering, sub/ses iteration, and the discover-process-export
-loop. CLI modules delegate to these after parsing arguments.
+Provides ``run()`` entry points for each workflow that handle runner setup,
+BIDS table loading, filtering, sub/ses iteration, and the
+discover-process-export loop. CLI modules delegate to these after parsing
+arguments.
 """
 
 from __future__ import annotations
 
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING
 
+from rbc.core import CPAC_ANTS_SEED
+from rbc.core.niwrap import setup_runner
+
 if TYPE_CHECKING:
     from collections.abc import Sequence
+    from pathlib import Path
+    from typing import Literal
+
+_DEFAULT_ENV_VARS = {
+    "ITK_GLOBAL_DEFAULT_NUMBER_OF_THREADS": "1",
+    "ANTS_RANDOM_SEED": CPAC_ANTS_SEED,
+}
 
 
 @dataclass(frozen=True)
@@ -27,3 +38,30 @@ class Filters:
     participant_label: Sequence[str] = field(default_factory=tuple)
     session_label: Sequence[str] = field(default_factory=tuple)
     task: str | None = None
+
+
+@dataclass(frozen=True)
+class RunnerConfig:
+    """Configuration for the execution backend.
+
+    Attributes:
+        runner: Execution backend (local, docker, podman, singularity).
+        verbose: Verbosity level.
+        tmp_dir: Temporary directory for intermediate files.
+    """
+
+    runner: Literal["auto", "local", "docker", "podman", "singularity"] = "local"
+    verbose: bool = False
+    tmp_dir: Path | None = None
+
+
+def init_runner(config: RunnerConfig) -> None:
+    """Set up the execution backend and environment variables.
+
+    Args:
+        config: Runner configuration.
+    """
+    ctx = setup_runner(
+        runner=config.runner, verbose=config.verbose, tmp_dir=config.tmp_dir
+    )
+    ctx.runner.environ = _DEFAULT_ENV_VARS

src/rbc/orchestration/all.py

@@ -17,6 +17,7 @@
 from rbc.bids.qc import export_qc
 from rbc.bids.session import load_session
 from rbc.context import RunContext
+from rbc.orchestration import Filters, RunnerConfig, init_runner
 from rbc.orchestration.anatomical import process_session as process_anat
 from rbc.orchestration.functional import process_session as process_func
 from rbc.workflows.metrics import single_session_metrics
@@ -26,7 +27,6 @@
     from collections.abc import Sequence
     from pathlib import Path
 
-    from rbc.orchestration import Filters
     from rbc_resources import AtlasName
 
 _logger = logging.getLogger(__name__)
@@ -42,7 +42,7 @@ def run(
     fwhm: float,
     start_tr: int,
     tr: float | None = None,
-    verbose: bool = False,
+    runner_config: RunnerConfig | None = None,
 ) -> None:
     """Run the full pipeline (anat + func + metrics + QC) per session.
 
@@ -58,8 +58,13 @@ def run(
         fwhm: Smoothing kernel FWHM in mm.
         start_tr: Number of initial TRs discarded during preprocessing.
         tr: TR override in seconds, or ``None`` to read from headers.
-        verbose: Show progress bar.
+        runner_config: Execution backend configuration.
     """
+    config = runner_config or RunnerConfig()
+    init_runner(config)
+    verbose = config.verbose
+
+    _logger.info("Preparing to run RBC full pipeline")
     df = load_table(
         dataset_dir=input_dir, index_fpath=None, max_workers=0, verbose=verbose
     )
@@ -140,3 +145,5 @@ def run(
             _logger.info("QC %s for sub-%s", status, pipe_ctx.sub)
 
         pipe_ctx.ensure_dataset_description()
+
+    _logger.info("RBC full pipeline complete")

src/rbc/orchestration/anatomical.py

@@ -12,13 +12,13 @@
 from rbc.bids.anatomical import discover_anatomical, export_anatomical
 from rbc.bids.session import load_session
 from rbc.context import RunContext
+from rbc.orchestration import Filters, RunnerConfig, init_runner
 from rbc.workflows.anatomical import AnatomicalOutputs, single_session_preprocess
 
 if TYPE_CHECKING:
     from pathlib import Path
 
     from rbc.bids.session import SessionTables
-    from rbc.orchestration import Filters
 
 _logger = logging.getLogger(__name__)
 
@@ -54,16 +54,21 @@ def run(
     output_dir: Path,
     *,
     filters: Filters,
-    verbose: bool = False,
+    runner_config: RunnerConfig | None = None,
 ) -> None:
     """Run the anatomical pipeline for all matching subjects/sessions.
 
     Args:
         input_dir: BIDS dataset directory.
         output_dir: Output directory for derivatives.
         filters: Participant/session/task filters.
-        verbose: Show progress bar.
+        runner_config: Execution backend configuration.
     """
+    config = runner_config or RunnerConfig()
+    init_runner(config)
+    verbose = config.verbose
+
+    _logger.info("Preparing to run RBC anatomical workflow")
     df = load_table(
         dataset_dir=input_dir, index_fpath=None, max_workers=0, verbose=verbose
     )
@@ -90,3 +95,5 @@ def run(
         session = load_session(sub_ses_group, pipe_ctx.sub, pipe_ctx.ses)
         process_session(session, pipe_ctx)
         pipe_ctx.ensure_dataset_description()
+
+    _logger.info("RBC anatomical workflow complete")