"""Input layout discovery for MRSIPrep."""
from __future__ import annotations
import json
import re
from dataclasses import dataclass
from pathlib import Path
from mrsiprep.config.defaults import METABOLITE_ALIASES
from mrsiprep.utils.misc import normalize_session, normalize_subject, parse_bids_entities
# Long-form PyBIDS/fMRIPrep-style entity names accepted in --bids-filter-file,
# mapped to the short-form keys parse_bids_entities() actually returns.
_FILTER_ENTITY_ALIASES = {
"subject": "sub",
"session": "ses",
"acquisition": "acq",
"run": "run",
"acq": "acq",
"sub": "sub",
"ses": "ses",
}
_SUPPORTED_BIDS_FILTER_KEYS = {"t1w"}
[docs]
@dataclass(frozen=True)
class Recording:
subject: str
session: str | None
@property
def prefix(self) -> str:
if self.session:
return f"sub-{self.subject}_ses-{self.session}"
return f"sub-{self.subject}"
[docs]
def load_bids_filters(path: str | Path | None) -> dict:
"""Parse a --bids-filter-file JSON document.
Only the "t1w" top-level key is currently supported (the one input
MRSIPrep has ambiguous-candidate logic for today, in
BIDSLayout.raw_t1()). Any other top-level key raises immediately rather
than being silently ignored, since a user copying an fMRIPrep-style
filter file (with "bold"/"fmap"/etc keys) should get fast feedback that
those aren't supported here yet.
"""
if path is None:
return {}
path = Path(path)
try:
data = json.loads(path.read_text())
except (OSError, json.JSONDecodeError) as exc:
raise ValueError(f"Could not parse --bids-filter-file {path}: {exc}") from exc
if not isinstance(data, dict):
raise ValueError(f"--bids-filter-file {path} must contain a JSON object at the top level.")
unsupported = set(data) - _SUPPORTED_BIDS_FILTER_KEYS
if unsupported:
raise ValueError(
f"--bids-filter-file {path} has unsupported key(s) {sorted(unsupported)}; "
f"only {sorted(_SUPPORTED_BIDS_FILTER_KEYS)} is currently supported."
)
return data
[docs]
class BIDSLayout:
"""Minimal path resolver for the MRSI-Metabolic-Connectome derivative layout."""
def __init__(self, bids_dir: str | Path, filters: dict | None = None):
self.bids_dir = Path(bids_dir).resolve()
self.derivatives = self.bids_dir / "derivatives"
self.filters = filters or {}
[docs]
def discover_recordings(self) -> list[Recording]:
participants = self.bids_dir / "participants_allsessions.tsv"
if participants.exists():
from mrsiprep.utils.misc import read_participant_pairs
return [Recording(sub, ses) for sub, ses in read_participant_pairs(participants)]
out: list[Recording] = []
for sub_dir in sorted(self.bids_dir.glob("sub-*")):
if not sub_dir.is_dir():
continue
subject = normalize_subject(sub_dir.name)
ses_dirs = sorted(sub_dir.glob("ses-*"))
if not ses_dirs:
out.append(Recording(subject, None))
for ses_dir in ses_dirs:
out.append(Recording(subject, normalize_session(ses_dir.name)))
return out
[docs]
def raw_t1(self, subject: str, session: str | None, reference_name: str | None = None) -> Path | None:
anat_dir = self._raw_anat_dir(subject, session)
if not anat_dir.exists():
return None
t1w_filter = self.filters.get("t1w")
candidates = []
for path in sorted(anat_dir.glob("*T1w.nii*")):
if "_desc-" in path.name:
continue
entities = parse_bids_entities(path)
if t1w_filter and not _matches_filter(entities, t1w_filter):
continue
if reference_name and reference_name in path.name:
return path
score = 0
if entities.get("acq") in {"memprage", "mprage", "mp2rage"}:
score += 2
if entities.get("run") == "01":
score += 1
candidates.append((score, path))
if not candidates:
return None
candidates.sort(key=lambda item: (-item[0], item[1].name))
return candidates[0][1]
[docs]
def t1(self, subject: str, session: str | None, pattern: str = "desc-brain_T1w") -> Path | None:
if pattern and Path(pattern).exists():
return Path(pattern).resolve()
roots = [
self.derivatives / "skullstrip" / f"sub-{normalize_subject(subject)}" / self._ses_dir(session),
self._raw_anat_dir(subject, session),
]
for root in roots:
if not root.exists():
continue
matches = sorted(path for path in root.glob("*.nii*") if pattern in path.name)
if matches:
return matches[0]
return self.raw_t1(subject, session)
[docs]
def brain_mask(self, subject: str, session: str | None) -> Path | None:
root = self.derivatives / "skullstrip" / f"sub-{normalize_subject(subject)}" / self._ses_dir(session)
if not root.exists():
return None
patterns = ["*desc-brainmask*T1w.nii*", "*desc-brain_mask.nii*", "*desc-brainmask.nii*"]
for pattern in patterns:
matches = sorted(root.glob(pattern))
if matches:
return matches[0]
return None
[docs]
def cat12_probseg(self, subject: str, session: str | None, index: int) -> Path | None:
root = self.derivatives / "cat12" / f"sub-{normalize_subject(subject)}" / self._ses_dir(session)
if not root.exists():
return None
matches = sorted(root.glob(f"*desc-p{index}_T1w.nii*"))
return matches[0] if matches else None
def _mrsi_input_roots(self, subject: str, session: str | None, space: str) -> list[Path]:
"""Candidate roots for raw MRSI inputs, in priority order.
The primary convention is a top-level `derivatives/mrsi-<space>/`
tree (used by real acquisition datasets). When that root is missing
or empty for this subject/session, fall back to MRSIPrep's own
output layout (`derivatives/mrsiprep/sub-*/ses-*/mrsi/<space>/`) --
this lets synthetic/test datasets that pre-populate MRSIPrep's
output-shaped tree (e.g. synthMRSI) be consumed directly as input
without a separate mrsi-<space> mirror.
"""
sub = f"sub-{normalize_subject(subject)}"
ses = self._ses_dir(session)
primary = self.derivatives / f"mrsi-{space}" / sub / ses
roots = [primary]
if not primary.exists() or not any(primary.iterdir()):
fallback = self.derivatives / "mrsiprep" / sub / ses / "mrsi" / space
roots.append(fallback)
return roots
[docs]
def mrsi_map(
self,
subject: str,
session: str | None,
desc: str,
met: str | None = None,
option: str | None = None,
space: str = "orig",
res: str | int | None = None,
construct: bool = False,
) -> Path | None:
roots = self._mrsi_input_roots(subject, session, space)
filename = self._mrsi_filename(subject, session, desc, met, option, space, res)
for root in roots:
direct = root / filename
if direct.exists() or (construct and root is roots[0]):
return direct
if met:
for alias in METABOLITE_ALIASES.get(met, [met]):
alias_filename = self._mrsi_filename(subject, session, desc, alias, option, space, res)
path = root / alias_filename
if path.exists():
return path
if not root.exists():
continue
regex = self._mrsi_regex(subject, session, desc, met, option, space, res)
matches = sorted(path for path in root.glob("*.nii*") if regex.match(path.name))
if matches:
return matches[0]
return None
[docs]
def chimera_atlas(self, subject: str, session: str | None, scheme: str, scale: int, grow: int = 2, space: str = "orig") -> Path | None:
root = self.derivatives / "chimera-atlases" / f"sub-{normalize_subject(subject)}" / self._ses_dir(session) / "anat"
if not root.exists():
return None
space_token = "orig" if space.lower() in {"t1w", "anat"} else space
pattern = f"{self._prefix(subject, session)}*space-{space_token}_atlas-chimera{scheme}_desc-scale{scale}grow{grow}mm_dseg.nii*"
matches = sorted(root.glob(pattern))
return matches[0] if matches else None
def _raw_anat_dir(self, subject: str, session: str | None) -> Path:
path = self.bids_dir / f"sub-{normalize_subject(subject)}"
ses = normalize_session(session)
if ses:
path = path / f"ses-{ses}"
return path / "anat"
def _ses_dir(self, session: str | None) -> str:
ses = normalize_session(session)
return f"ses-{ses}" if ses else ""
def _prefix(self, subject: str, session: str | None) -> str:
sub = f"sub-{normalize_subject(subject)}"
ses = normalize_session(session)
return f"{sub}_ses-{ses}" if ses else sub
def _mrsi_filename(self, subject: str, session: str | None, desc: str, met: str | None, option: str | None, space: str, res: str | int | None) -> str:
parts = [self._prefix(subject, session), f"space-{space}"]
if res is not None:
res_str = str(res)
if not res_str.endswith("mm"):
res_str = f"{res_str}mm"
parts.append(f"res-{res_str}")
if met:
parts.append(f"met-{met}")
parts.append(f"desc-{desc}")
if option:
parts.append(str(option))
return "_".join(parts) + "_mrsi.nii.gz"
def _mrsi_regex(self, subject: str, session: str | None, desc: str, met: str | None, option: str | None, space: str, res: str | int | None):
prefix = re.escape(self._prefix(subject, session))
met_part = rf"_met-{re.escape(met)}" if met else ""
res_part = r"_res-[^_]+" if res is not None else r"(?:_res-[^_]+)?"
option_part = rf"_{re.escape(option)}" if option else r"(?:_[^_]+)?"
return re.compile(rf"^{prefix}_space-{re.escape(space)}{res_part}{met_part}_desc-{re.escape(desc)}{option_part}_mrsi\.nii(\.gz)?$")
def _matches_filter(entities: dict, filter_dict: dict) -> bool:
"""True if every key in filter_dict matches the corresponding parsed entity.
Filter keys may use either MRSIPrep's short entity names (acq, run, ses,
sub) or PyBIDS/fMRIPrep-style long names (acquisition, session, subject),
per _FILTER_ENTITY_ALIASES. A filter value of null/None requires the
entity to be absent from the filename.
"""
for key, expected in filter_dict.items():
short_key = _FILTER_ENTITY_ALIASES.get(key, key)
actual = entities.get(short_key)
if expected is None:
if actual is not None:
return False
elif actual != str(expected):
return False
return True