Skip to content

feat: add opt-in literal output layout #144

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
viranovskaya wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

feat: add opt-in literal output layout #144

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 35 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -145,6 +145,41 @@ derivatives/MEEGqc/
summary_reports/ GQI artefacts (TSV + JSON), versioned via attempt files
```

An external output keeps the same BIDS-style structure by default:

```bash
run-meegqc --inputdata /path/to/dataset \
--derivatives_output /path/to/output
# /path/to/output/dataset/derivatives/MEEGqc/
```

Use the opt-in literal layout when the selected folder should contain
`MEEGqc` directly:

```bash
run-meegqc --inputdata /path/to/dataset \
--derivatives_output /path/to/output \
--output_layout literal
# /path/to/output/MEEGqc/
```

For multiple input datasets, literal layout adds one dataset-named folder to
keep their outputs separate: `/path/to/output/<dataset>/MEEGqc/`.

The Python API uses the same option:

```python
from meg_qc.calculation.meg_qc_pipeline import make_derivative_meg_qc

make_derivative_meg_qc(
default_config_file_path="/path/to/settings.ini",
internal_config_file_path="/path/to/settings_internal.ini",
ds_paths="/path/to/dataset",
derivatives_base="/path/to/output",
output_layout="literal",
)
```

The GQI is written per-modality:
`group_metrics/meg/Global_Quality_Index_attempt_<n>_meg.tsv` and
`group_metrics/eeg/Global_Quality_Index_attempt_<n>_eeg.tsv`.
Expand Down
138 changes: 63 additions & 75 deletions meg_qc/calculation/meg_qc_pipeline.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -120,9 +120,14 @@ def _safe_map_object(self, model_type, json_object, target=None):
import json
import pandas as pd
from typing import Union, Optional, Dict, Tuple
from contextlib import contextmanager

from meg_qc.calculation.metrics.summary_report_GQI import generate_gqi_summary
from meg_qc.output_paths import (
dataset_derivatives_output,
derivative_write_context,
normalize_output_layout,
resolve_output_roots,
)


# Analysis-mode names. The top-level choice is "non-profile" (write straight to
Expand Down Expand Up @@ -159,56 +164,6 @@ def _timestamp_analysis_id() -> str:
return dt.datetime.now().strftime("%Y%m%d_%H%M%S")


def resolve_output_roots(dataset_path: str, external_derivatives_root: Optional[str]) -> Tuple[str, str]:
"""Return the dataset output root and derivatives folder respecting overrides.

Parameters
----------
dataset_path : str
Path to the original BIDS dataset.
external_derivatives_root : Optional[str]
User-provided folder in which a dataset-named directory will be created
to host derivatives. If ``None`` the derivatives live inside the
original dataset.

Returns
-------
tuple
``(output_root, derivatives_root)`` where ``output_root`` is the base
dataset directory used when writing derivatives and ``derivatives_root``
points to the "derivatives" folder inside ``output_root``.
"""

ds_name = os.path.basename(os.path.normpath(dataset_path))
output_root = dataset_path if external_derivatives_root is None else os.path.join(external_derivatives_root, ds_name)
derivatives_root = os.path.join(output_root, 'derivatives')
os.makedirs(derivatives_root, exist_ok=True)

# When output is external, seed output_root with a dataset_description.json.
# The plotting module loads ANCPBIDS directly from output_root (no symlink
# overlay) so this file must be present for schema-version detection.
if external_derivatives_root is not None:
desc_dst = os.path.join(output_root, "dataset_description.json")
if not os.path.exists(desc_dst):
desc_src = os.path.join(dataset_path, "dataset_description.json")
if os.path.exists(desc_src):
try:
import shutil as _shutil
_shutil.copy2(desc_src, desc_dst)
except OSError:
pass
if not os.path.exists(desc_dst):
import json as _json
stub = {"Name": os.path.basename(output_root), "BIDSVersion": "1.8.0"}
try:
with open(desc_dst, 'w', encoding='utf-8') as _fh:
_json.dump(stub, _fh, indent=2)
except OSError:
pass # Non-fatal; ancpbids falls back gracefully.

return output_root, derivatives_root


def _scan_profile_dir(profiles_root: str) -> List[Tuple[str, float]]:
"""Scan one profiles directory and return ``(name, mtime)`` pairs sorted by mtime desc.

Expand All @@ -232,6 +187,7 @@ def _scan_profile_dir(profiles_root: str) -> List[Tuple[str, float]]:
def list_analysis_profiles(
dataset_path: str,
external_derivatives_root: Optional[str] = None,
output_layout: str = "bids",
) -> List[str]:
"""List available MEGqc profile IDs for one dataset.

Expand All @@ -249,7 +205,11 @@ def list_analysis_profiles(
output path. Without this dual search the GUI "Load profiles" dialog shows
nothing and ``resolve_analysis_root`` raises ``FileNotFoundError``.
"""
_, derivatives_root = resolve_output_roots(dataset_path, external_derivatives_root)
_, derivatives_root = resolve_output_roots(
dataset_path,
external_derivatives_root,
output_layout=output_layout,
)
primary_profiles_root = os.path.join(derivatives_root, "MEEGqc", "profiles")
candidates = _scan_profile_dir(primary_profiles_root)

Expand Down Expand Up @@ -283,14 +243,19 @@ def list_analysis_profiles(
def has_only_legacy_derivatives(
dataset_path: str,
external_derivatives_root: Optional[str] = None,
output_layout: str = "bids",
) -> bool:
"""Return True if old-style ('Meg_QC') derivatives exist but no new ('MEEGqc') ones.

Lets the CLI and GUI show a clear, actionable message when a user tries to
build reports from derivatives produced by a pre-BIDS-rename MEEGqc version.
"""
try:
_output_root, derivatives_root = resolve_output_roots(dataset_path, external_derivatives_root)
_output_root, derivatives_root = resolve_output_roots(
dataset_path,
external_derivatives_root,
output_layout=output_layout,
)
except Exception:
return False
bases = [derivatives_root]
Expand All @@ -308,6 +273,7 @@ def resolve_analysis_root(
analysis_mode: str = "non-profile",
analysis_id: Optional[str] = None,
create_if_missing: bool = False,
output_layout: str = "bids",
) -> Tuple[str, str, str, Optional[str], List[str]]:
"""Resolve output roots plus profile-specific MEGqc folder.

Expand All @@ -325,7 +291,11 @@ def resolve_analysis_root(
f"{', '.join(sorted(_ANALYSIS_MODES))}."
)

output_root, derivatives_root = resolve_output_roots(dataset_path, external_derivatives_root)
output_root, derivatives_root = resolve_output_roots(
dataset_path,
external_derivatives_root,
output_layout=output_layout,
)
non_profile_root = os.path.join(derivatives_root, "MEEGqc")
if mode == "non-profile":
if create_if_missing:
Expand All @@ -337,7 +307,11 @@ def resolve_analysis_root(
os.makedirs(profiles_root, exist_ok=True)

resolved_id = analysis_id.strip() if isinstance(analysis_id, str) and analysis_id.strip() else None
available = list_analysis_profiles(dataset_path, external_derivatives_root)
available = list_analysis_profiles(
dataset_path,
external_derivatives_root,
output_layout=output_layout,
)

if mode == "new-profile":
resolved_id = resolved_id or _timestamp_analysis_id()
Expand Down Expand Up @@ -411,22 +385,6 @@ def _resolve_config_by_policy(
return default_config_file_path


@contextmanager
def temporary_dataset_base(dataset, base_dir: str):
"""Temporarily point an ANCPBIDS dataset to a different base directory.

This is used to redirect derivative writing without interfering with how
raw files are located inside the original BIDS dataset.
"""

original_base = getattr(dataset, 'base_dir_', None)
dataset.base_dir_ = base_dir
try:
yield
finally:
dataset.base_dir_ = original_base


def _ensure_derivative_dataset_description_filename(derivative) -> None:
"""Guarantee a writable dataset_description filename for ANCPBIDS writes.

Expand Down Expand Up @@ -1191,6 +1149,7 @@ def process_one_subject(
derivatives_root: str,
output_root: str,
analysis_segments: Optional[List[str]] = None,
output_layout: str = "bids",
):
"""
This function processes a single subject. It contains all the code that was
Expand All @@ -1213,6 +1172,8 @@ def process_one_subject(
output_root : str
Base directory used when persisting derivatives (parent of the
derivatives folder), allowing redirection outside the BIDS dataset.
output_layout : str
Output tree convention used when persisting ANCPBIDS derivatives.
analysis_segments : list of str, optional
Extra folder segments inserted between ``MEEGqc`` and ``calculation``.
In legacy mode this is ``[]``; in profile mode this is
Expand Down Expand Up @@ -1684,7 +1645,12 @@ def json_writer(file_path, cont=deriv.content):
print('REMOVING TRASH: FAILED')

# WRITE DERIVATIVE
with temporary_dataset_base(dataset, output_root):
with derivative_write_context(
dataset,
derivative,
output_root,
output_layout,
):
ancpbids.write_derivative(dataset, derivative)

# Removes intermediate trash objects — guard against NameError when no
Expand Down Expand Up @@ -1717,7 +1683,8 @@ def process_one_subject_safe(
internal_qc_params: dict,
derivatives_root: str,
output_root: str,
analysis_segments: Optional[List[str]] = None):
analysis_segments: Optional[List[str]] = None,
output_layout: str = "bids"):
"""Wrapper around :func:`process_one_subject` that catches errors.

Parameters are identical to :func:`process_one_subject`.
Expand Down Expand Up @@ -1749,6 +1716,7 @@ def process_one_subject_safe(
internal_qc_params=internal_qc_params,
derivatives_root=derivatives_root,
output_root=output_root,
output_layout=output_layout,
analysis_segments=analysis_segments,
)
# process_one_subject now returns (files, metric_errors)
Expand Down Expand Up @@ -1923,6 +1891,7 @@ def make_derivative_meg_qc(
processed_subjects_policy: str = "skip",
interactive_prompts: bool = False,
keep_temp_on_error: bool = False,
output_layout: str = "bids",
):
"""Run MEGqc calculation for one or more datasets.

Expand All @@ -1943,15 +1912,27 @@ def make_derivative_meg_qc(
keep_temp_on_error
If ``True``, temporary preprocessed FIF files are kept when an exception
occurs while processing a dataset to aid debugging.
output_layout
``"bids"`` keeps the existing external layout
``<output>/<dataset>/derivatives/MEEGqc``. ``"literal"`` writes to
``<output>/MEEGqc`` and requires ``derivatives_base``.
"""
start_time = time.time()

ds_paths = check_ds_paths(ds_paths)
output_layout = normalize_output_layout(output_layout)
internal_qc_params = get_internal_config_params(internal_config_file_path)

requested_sub_list = sub_list
multiple_datasets = len(ds_paths) > 1

for dataset_path in ds_paths:
dataset_derivatives_base = dataset_derivatives_output(
derivatives_base,
dataset_path,
output_layout,
multiple_datasets=multiple_datasets,
)
print('___MEGqc___: ', 'DS path:', dataset_path)
dataset = ancpbids.load_dataset(dataset_path, DatasetOptions(lazy_loading=True))
(
Expand All @@ -1962,7 +1943,8 @@ def make_derivative_meg_qc(
analysis_segments,
) = resolve_analysis_root(
dataset_path=dataset_path,
external_derivatives_root=derivatives_base,
external_derivatives_root=dataset_derivatives_base,
output_layout=output_layout,
analysis_mode=analysis_mode,
analysis_id=analysis_id,
create_if_missing=True,
Expand Down Expand Up @@ -2020,6 +2002,7 @@ def make_derivative_meg_qc(
internal_qc_params=internal_qc_params,
derivatives_root=megqc_root,
output_root=output_root,
output_layout=output_layout,
analysis_segments=analysis_segments,
)
for sub in dataset_sub_list
Expand Down Expand Up @@ -2080,7 +2063,12 @@ def make_derivative_meg_qc(
create_config_artifact(root_folder, config_file_path, 'UsedSettings', all_subs_raw_files)

# Write the pipeline-level derivative to disk
with temporary_dataset_base(dataset, output_root):
with derivative_write_context(
dataset,
derivative,
output_root,
output_layout,
):
ancpbids.write_derivative(dataset, derivative)

_write_profile_manifest(
Expand Down
Loading