✨ Expose typed Python API for the seven wk-* operations

codecov.yml

@@ -0,0 +1,10 @@
+coverage:
+  status:
+    project:
+      default:
+        target: auto
+        threshold: 1%
+    patch:
+      default:
+        target: 90%
+        threshold: 1%

tests/test_scripts.py

@@ -600,6 +600,91 @@ def test_unwrap_phase_noiseframes_consumes_all_frames(argv, capsys, tmp_path):
     assert "0 frames" in err
 
 
+def test_unwrap_phase_metadata_accepts_echotime_only(argv, capsys, tmp_path):
+    """``wk-unwrap-phase`` only needs EchoTime; sidecars without
+    TotalReadoutTime / PhaseEncodingDirection must work. A mismatched phase
+    count forces a clean parser.error *after* the metadata loader resolves —
+    if the loader still required TRT/PED we'd see a KeyError instead."""
+    sidecar = tmp_path / "m1.json"
+    sidecar.write_text(json.dumps({"EchoTime": 0.014}))
+    argv(
+        [
+            "wk-unwrap-phase",
+            "--magnitude",
+            "m.nii",
+            "--phase",
+            "p1.nii",
+            "p2.nii",
+            "--metadata",
+            str(sidecar),
+            "--out-prefix",
+            str(tmp_path / "out"),
+        ]
+    )
+    with pytest.raises(SystemExit) as exc:
+        unwrap_phase_main()
+    assert exc.value.code == 2
+    err = capsys.readouterr().err
+    assert "must match" in err
+
+
+def test_unwrap_phase_metadata_missing_echotime(argv, capsys, tmp_path):
+    """A sidecar without EchoTime must surface as a clean parser error, not a
+    KeyError from the metadata loader."""
+    sidecar = tmp_path / "m1.json"
+    sidecar.write_text(json.dumps({}))
+    argv(
+        [
+            "wk-unwrap-phase",
+            "--magnitude",
+            "m.nii",
+            "--phase",
+            "p.nii",
+            "--metadata",
+            str(sidecar),
+            "--out-prefix",
+            str(tmp_path / "out"),
+        ]
+    )
+    with pytest.raises(SystemExit) as exc:
+        unwrap_phase_main()
+    assert exc.value.code == 2
+    err = capsys.readouterr().err
+    assert "EchoTime" in err
+
+
+def test_medic_noiseframes_consumes_all_frames(argv, capsys, tmp_path):
+    """``-f`` >= n_frames now raises a clean parser error in medic too: the
+    check moved into ``trim_noise_frames`` so every caller is protected from
+    silently producing an empty 4D series."""
+    mag = _write_nifti(tmp_path / "m.nii", (4, 4, 4, 5))
+    phase = _write_nifti(tmp_path / "p.nii", (4, 4, 4, 5))
+    argv(
+        [
+            "wk-medic",
+            "--magnitude",
+            mag,
+            "--phase",
+            phase,
+            "--TEs",
+            "14.0",
+            "--total-readout-time",
+            "0.05",
+            "--phase-encoding-direction",
+            "j",
+            "--out-prefix",
+            str(tmp_path / "out"),
+            "-f",
+            "5",
+        ]
+    )
+    with pytest.raises(SystemExit) as exc:
+        medic_main()
+    assert exc.value.code == 2
+    err = capsys.readouterr().err
+    assert "0 frames" in err
+
+
 # ---------------------------------------------------------------------------
 # compute_fieldmap --help / argument validation
 # ---------------------------------------------------------------------------
@@ -1693,14 +1778,12 @@ def test_bundle_frames_to_3d_series_clears_vector_intent():
 
 def test_write_output_per_frame_map_clears_vector_intent(tmp_path):
     """Per-frame map outputs must round-trip without a stale vector intent."""
-    import argparse
-
     from warpkit.scripts._warp_io import write_output
 
     frames = [_vector_intent_frame() for _ in range(2)]
     out_paths = [str(tmp_path / "f1.nii"), str(tmp_path / "f2.nii")]
 
-    write_output(frames, out_paths, "map", argparse.ArgumentParser())
+    write_output(frames, out_paths, "map")
 
     for p in out_paths:
         loaded = _load(p)

warpkit/api.py

@@ -0,0 +1,48 @@
+"""Typed Python entry points for the seven warpkit operations.
+
+Every ``wk-*`` CLI tool is mirrored here as a typed Python function so
+library callers (e.g. nipype interfaces, fmriprep) can drive warpkit without
+shelling out or fabricating ``argv``. Each function takes keyword-only
+arguments, raises :class:`ValueError` on validation problems, writes its
+outputs to disk, and returns a frozen dataclass with the absolute paths of
+the written NIfTIs.
+
+Mapping CLI -> Python:
+
+* ``wk-medic``              -> :func:`medic` -> :class:`MedicResult`
+* ``wk-unwrap-phase``       -> :func:`unwrap_phase` -> :class:`UnwrapPhaseResult`
+* ``wk-compute-fieldmap``   -> :func:`compute_fieldmap` -> :class:`ComputeFieldmapResult`
+* ``wk-apply-warp``         -> :func:`apply_warp` -> :class:`ApplyWarpResult`
+* ``wk-convert-warp``       -> :func:`convert_warp` -> :class:`ConvertWarpResult`
+* ``wk-convert-fieldmap``   -> :func:`convert_fieldmap` -> :class:`ConvertFieldmapResult`
+* ``wk-compute-jacobian``   -> :func:`compute_jacobian` -> :class:`ComputeJacobianResult`
+
+The CLI flag-name → Python kwarg mapping is the obvious dash-to-underscore
+transform; the only difference is ``--TEs`` (kept for MR convention) →
+``tes`` (lowercase, per repo style).
+"""
+
+from .scripts.apply_warp import ApplyWarpResult, apply_warp
+from .scripts.compute_fieldmap import ComputeFieldmapResult, compute_fieldmap
+from .scripts.compute_jacobian import ComputeJacobianResult, compute_jacobian
+from .scripts.convert_fieldmap import ConvertFieldmapResult, convert_fieldmap
+from .scripts.convert_warp import ConvertWarpResult, convert_warp
+from .scripts.medic import MedicResult, medic
+from .scripts.unwrap_phase import UnwrapPhaseResult, unwrap_phase
+
+__all__ = [
+    "ApplyWarpResult",
+    "ComputeFieldmapResult",
+    "ComputeJacobianResult",
+    "ConvertFieldmapResult",
+    "ConvertWarpResult",
+    "MedicResult",
+    "UnwrapPhaseResult",
+    "apply_warp",
+    "compute_fieldmap",
+    "compute_jacobian",
+    "convert_fieldmap",
+    "convert_warp",
+    "medic",
+    "unwrap_phase",
+]

warpkit/scripts/_metadata.py

@@ -0,0 +1,162 @@
+"""Shared acquisition-metadata helpers for the warpkit script entry points.
+
+Centralises:
+
+* coercing user-supplied images (``Path`` / ``str`` / ``Nifti1Image``) into
+  ``Nifti1Image`` objects,
+* loading echo time / total readout time / phase encoding direction from
+  BIDS-style JSON sidecars,
+* the "either ``--metadata`` or direct args" mutex/either-or check shared by
+  ``wk-medic``, ``wk-unwrap-phase``, and ``wk-compute-fieldmap``.
+
+Validation errors raise :class:`ValueError`; the CLI shims forward those to
+``parser.error`` so the user-visible behaviour (``SystemExit(2)``) is
+preserved.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Sequence
+from os import PathLike
+from typing import cast
+
+import nibabel as nib
+
+
+def ensure_image(x: PathLike[str] | str | nib.Nifti1Image) -> nib.Nifti1Image:
+    """Coerce a path or in-memory image into a ``Nifti1Image``."""
+    if isinstance(x, nib.Nifti1Image):
+        return x
+    return cast(nib.Nifti1Image, nib.load(str(x)))
+
+
+def ensure_images(
+    xs: Sequence[PathLike[str] | str | nib.Nifti1Image],
+) -> list[nib.Nifti1Image]:
+    return [ensure_image(x) for x in xs]
+
+
+def load_acquisition_from_metadata(
+    metadata_paths: Sequence[PathLike[str] | str],
+    *,
+    require_trt_pe: bool = True,
+) -> tuple[list[float], float | None, str | None]:
+    """Read EchoTime (s → ms) — and optionally TotalReadoutTime (s) and
+    PhaseEncodingDirection — from BIDS-style JSON sidecars.
+
+    Per-echo ``EchoTime`` is read from each file; ``TotalReadoutTime`` and
+    ``PhaseEncodingDirection`` are taken from the first. When
+    ``require_trt_pe=False`` the latter two are skipped so callers that only
+    need echo times (e.g. ``unwrap_phase``) accept sidecars that omit them.
+    Missing required keys raise :class:`ValueError`.
+    """
+    metadatas = []
+    for j in metadata_paths:
+        with open(j) as f:
+            metadatas.append(json.load(f))
+    try:
+        tes_ms = [float(m["EchoTime"]) * 1000 for m in metadatas]
+    except KeyError:
+        raise ValueError(
+            "metadata sidecar is missing required key: 'EchoTime'."
+        ) from None
+    if not require_trt_pe:
+        return tes_ms, None, None
+    missing = [
+        k
+        for k in ("TotalReadoutTime", "PhaseEncodingDirection")
+        if k not in metadatas[0]
+    ]
+    if missing:
+        raise ValueError(
+            "metadata sidecar is missing required key(s): "
+            f"{', '.join(repr(k) for k in missing)}."
+        )
+    trt = float(metadatas[0]["TotalReadoutTime"])
+    ped = str(metadatas[0]["PhaseEncodingDirection"])
+    return tes_ms, trt, ped
+
+
+def resolve_acquisition(
+    *,
+    metadata: Sequence[PathLike[str] | str] | None,
+    tes: Sequence[float] | None,
+    total_readout_time: float | None = None,
+    phase_encoding_direction: str | None = None,
+    require_trt_pe: bool = True,
+) -> tuple[list[float], float | None, str | None]:
+    """Resolve echo times / TRT / PED from either BIDS metadata or direct args.
+
+    Mirrors the either-or / mutex logic in the CLI scripts. Set
+    ``require_trt_pe=False`` for callers that only need echo times (e.g.
+    ``unwrap_phase``); the error message then matches that script's wording.
+
+    Error messages reference the dash-form CLI flags (``--metadata``,
+    ``--TEs``, ...) so CLI tests continue to match; nipype/library callers
+    will see the same text via ``ValueError``.
+    """
+    flag_map = {
+        "tes": "--TEs",
+        "total_readout_time": "--total-readout-time",
+        "phase_encoding_direction": "--phase-encoding-direction",
+    }
+    if require_trt_pe:
+        direct_vals = {
+            "tes": tes,
+            "total_readout_time": total_readout_time,
+            "phase_encoding_direction": phase_encoding_direction,
+        }
+    else:
+        direct_vals = {"tes": tes}
+    direct_supplied = [k for k, v in direct_vals.items() if v is not None]
+
+    if metadata is not None and direct_supplied:
+        names = ", ".join(flag_map[k] for k in direct_supplied)
+        raise ValueError(
+            f"--metadata is mutually exclusive with {names}; pass one or the "
+            "other, not both."
+        )
+    if metadata is None and len(direct_supplied) != len(direct_vals):
+        if require_trt_pe:
+            missing = [flag_map[k] for k in direct_vals if k not in direct_supplied]
+            raise ValueError(
+                "either --metadata or all of --TEs, --total-readout-time, and "
+                f"--phase-encoding-direction must be provided (missing: {', '.join(missing)})."
+            )
+        raise ValueError("either --metadata or --TEs must be provided.")
+
+    if metadata is not None:
+        return load_acquisition_from_metadata(metadata, require_trt_pe=require_trt_pe)
+
+    return list(tes or []), total_readout_time, phase_encoding_direction
+
+
+def trim_noise_frames(images: list[nib.Nifti1Image], n: int) -> list[nib.Nifti1Image]:
+    """Trim the last ``n`` frames from each 4D image. Returns the input list
+    unchanged when ``n == 0``.
+
+    When ``n > 0`` each image must be 4D with strictly more than ``n`` frames;
+    otherwise ``[..., :-n]`` would either chop the Z dimension of a 3D volume
+    or yield an empty 4D series that crashes downstream consumers. Both raise
+    :class:`ValueError`.
+    """
+    if n == 0:
+        return images
+    if n < 0:
+        raise ValueError(f"noise_frames must be non-negative; got {n}.")
+    for idx, img in enumerate(images):
+        if img.ndim != 4:
+            raise ValueError(
+                f"noise_frames={n} requires 4D images; image #{idx} has "
+                f"ndim={img.ndim}."
+            )
+        n_frames = img.shape[-1]
+        if n >= n_frames:
+            raise ValueError(
+                f"noise_frames={n} would leave 0 frames in image #{idx} "
+                f"(has {n_frames} frame(s))."
+            )
+    return [
+        nib.Nifti1Image(img.dataobj[..., :-n], img.affine, img.header) for img in images
+    ]