"""HCP path resolution and preparation."""
from pathlib import Path
from typing import TypedDict
from thesis.core.config import PipelineConfig
from thesis.core.context import ProcessingContext
from thesis.core.logging import get_logger
from thesis.core.utils import resolve_path
from ..common import format_patient_path, resolve_t1_path, resolve_with_fallback
from ..config.values import resolve_hcp_value
logger = get_logger(__name__)
[docs]
class HCPPaths(TypedDict):
"""Resolved filesystem paths for the HCP workflow."""
input_dir: Path
diffusion_dir: Path
bedpostx_dir: Path
samples_base_name: Path
thsamples: list[Path]
phsamples: list[Path]
fsamples: list[Path]
mask_path: Path
t1_path: Path
[docs]
def prepare_hcp_paths(config: PipelineConfig, context: ProcessingContext) -> HCPPaths:
"""Resolve HCP directory paths from config and context."""
diffusion_dirname = format_patient_path(
resolve_hcp_value(config, "diffusion_dir", "T1w/Diffusion"), context.patient_id
)
bedpostx_dirname = format_patient_path(
resolve_hcp_value(config, "bedpostx_dir", "T1w/Diffusion.bedpostX"), context.patient_id
)
input_dir = (
Path(context.input_dir).resolve() if context.input_dir is not None else Path(".").resolve()
)
diffusion_dir = resolve_path(input_dir, diffusion_dirname)
bedpostx_dir = resolve_path(input_dir, bedpostx_dirname)
mask_name = resolve_hcp_value(config, "mask_name", "nodif_brain_mask.nii")
mask_path_cfg = resolve_hcp_value(config, "mask_path")
if mask_path_cfg:
mask_path = resolve_with_fallback(
format_patient_path(mask_path_cfg, context.patient_id),
input_dir,
[context.output_dir, context.data_dir],
)
else:
mask_path = diffusion_dir / mask_name
if not mask_path.exists():
for fb in (input_dir, Path("."), getattr(context, "data_dir", None)):
if fb is None:
continue
fb_path = Path(fb).resolve() / mask_name
if fb_path.exists():
logger.warning(
"Brain mask not found at {}, using fallback: {}",
diffusion_dir / mask_name,
fb_path,
)
mask_path = fb_path
break
# BedpostX sample discovery: scan primary dir then non-recursive/recursive
# fallbacks. Stops at the first dir with matching theta samples; the ph/f
# samples are then read from the same directory.
sample_pattern = "merged_th*samples.nii.gz"
thsamples = sorted(bedpostx_dir.glob(sample_pattern))
found_dir = bedpostx_dir
searched: list[Path] = [bedpostx_dir]
if not thsamples:
for fb in (input_dir, context.output_dir, getattr(context, "data_dir", None), Path(".")):
if fb is None:
continue
fb_path = Path(fb).resolve()
if fb_path == bedpostx_dir.resolve() or fb_path in searched:
continue
searched.append(fb_path)
found = sorted(fb_path.glob(sample_pattern)) or sorted(fb_path.rglob(sample_pattern))
if found:
thsamples = found
found_dir = found[0].parent
logger.warning(
"BedpostX samples not found in {}, using fallback: {}", bedpostx_dir, found_dir
)
break
if not thsamples:
logger.warning(
"No BedpostX theta samples ({}) found under: {}. Run BedpostX or "
"set hcp.bedpostx_dir in your config.",
sample_pattern,
", ".join(str(d) for d in searched),
)
return {
"input_dir": input_dir,
"diffusion_dir": diffusion_dir,
"bedpostx_dir": bedpostx_dir,
"samples_base_name": found_dir / "merged",
"thsamples": thsamples,
"phsamples": sorted(found_dir.glob("merged_ph*samples.nii.gz")),
"fsamples": sorted(found_dir.glob("merged_f*samples.nii.gz")),
"mask_path": mask_path,
"t1_path": resolve_t1_path(config, context),
}
[docs]
def prepare_seed_path(config: PipelineConfig, input_dir: Path) -> Path:
"""Resolve seed ROI path from config.
Defaults to ``<diffusion_dir>/seed_source.nii.gz`` with input_dir/cwd
fallback when ``tractography.seed_roi`` is not configured.
"""
seed_roi = getattr(getattr(config, "tractography", None), "seed_roi", None)
if seed_roi is not None:
patient_id = getattr(config, "patient_id", "") or ""
return resolve_path(input_dir, format_patient_path(str(seed_roi), patient_id))
diffusion_dir = resolve_path(
input_dir, resolve_hcp_value(config, "diffusion_dir", "T1w/Diffusion")
)
seed_path = diffusion_dir / "seed_source.nii.gz"
if seed_path.exists():
return seed_path
for fb in (input_dir, Path(".")):
alt = fb / "seed_source.nii.gz"
if alt.exists():
logger.warning("Seed image not found at {}, using fallback: {}", seed_path, alt)
return alt
return seed_path
