REF: Generalize bvals and uptake value iterators into a single iterator

Generalize bvals and uptake value iterators into a single iterator that
is able to traverse the values in ascending or descending order
depending on whether it is provided a list of b-values (DWI) or uptake
values (PET).

Follow-up to commit a1310e6.

Author: jhlegarreta

src/nifreeze/utils/iterators.py

@@ -24,7 +24,7 @@
 
 import random
 from itertools import chain, zip_longest
-from typing import Iterator
+from typing import Iterator, Sequence
 
 DEFAULT_ROUND_DECIMALS = 2
 """Round decimals to use when comparing values to be sorted for iteration purposes."""
@@ -172,14 +172,14 @@ def random_iterator(**kwargs) -> Iterator[int]:
 
 
 def _value_iterator(
-    values: list, ascending: bool, round_decimals: int = DEFAULT_ROUND_DECIMALS
+    values: Sequence[float], ascending: bool, round_decimals: int = DEFAULT_ROUND_DECIMALS
 ) -> Iterator[int]:
     """
     Traverse the given values in ascending or descenting order.
 
     Parameters
     ----------
-    values : :obj:`list`
+    values : :obj:`Sequence`
         List of values to traverse.
     ascending : :obj:`bool`
         If ``True``, traverse in ascending order; traverse in descending order
@@ -211,70 +211,57 @@ def _value_iterator(
     return (index[1] for index in indexed_vals)
 
 
-def bvalue_iterator(*_, **kwargs) -> Iterator[int]:
-    """
-    Traverse the volumes in a DWI dataset by increasing b-value.
-
-    Parameters
-    ----------
-    bvals : :obj:`list`
-        List of b-values corresponding to all orientations of the dataset.
-        Please note that ``bvals`` is a keyword argument and MUST be provided
-        to generate the volume sequence.
-
-    Yields
-    ------
-    :obj:`int`
-        The next index.
+def monotonic_value_iterator(*_, **kwargs) -> Iterator[int]:
+    try:
+        feature = next(k for k in (BVALS_KWARG, UPTAKE_KWARG) if kwargs.get(k) is not None)
+    except StopIteration:
+        raise TypeError(KWARG_ERROR_MSG.format(kwarg=f"{BVALS_KWARG} or {UPTAKE_KWARG}"))
 
-    Examples
-    --------
-    >>> list(bvalue_iterator(bvals=[0.0, 0.0, 1000.0, 1000.0, 700.0, 700.0, 2000.0, 2000.0, 0.0]))
-    [0, 1, 8, 4, 5, 2, 3, 6, 7]
-
-    """
-    bvals = kwargs.pop(BVALS_KWARG, None)
-    if bvals is None:
-        raise TypeError(KWARG_ERROR_MSG.format(kwarg=BVALS_KWARG))
+    ascending = feature == BVALS_KWARG
+    values = kwargs[feature]
     return _value_iterator(
-        bvals, ascending=True, round_decimals=kwargs.pop("round_decimals", DEFAULT_ROUND_DECIMALS)
+        values,
+        ascending=ascending,
+        round_decimals=kwargs.get("round_decimals", DEFAULT_ROUND_DECIMALS),
     )
 
 
-def uptake_iterator(*_, **kwargs) -> Iterator[int]:
-    """
-    Traverse the volumes in a PET dataset by decreasing uptake value.
+monotonic_value_iterator.__doc__ = f"""
+Traverse the volumes by increasing b-value in a DWI dataset or by decreasing
+uptake value in a PET dataset.
 
-    This function assumes that each uptake value corresponds to a single volume,
-    and that this value summarizes the uptake of the volume in a meaningful way,
-    e.g. a mean value across the entire volume.
+This function requires ``bvals`` or ``uptake`` be a keyword argument to generate
+the volume sequence. The b-values are assumed to all orientations in a DWI
+dataset, and uptake uptake values correspond to all volumes in a PET dataset.
 
-    Parameters
-    ----------
-    uptake : :obj:`list`
-        List of uptake values corresponding to all volumes of the dataset.
-        Please note that ``uptake`` is a keyword argument and MUST be provided
-        to generate the volume sequence.
+It is assumed that each uptake value corresponds to a single volume, and that
+this value summarizes the uptake of the volume in a meaningful way, e.g. a mean
+value across the entire volume.
 
-    Yields
-    ------
-    :obj:`int`
-        The next index.
+Other Parameters
+----------------
+{SIZE_KEYS_DOC}
 
-    Examples
-    --------
-    >>> list(uptake_iterator(uptake=[-1.23, 1.06, 1.02, 1.38, -1.46, -1.12, -1.19, 1.24, 1.05]))
-    [3, 7, 1, 8, 2, 5, 6, 0, 4]
+Notes
+-----
+Only one of the above keyword arguments may be provided at a time. If ``size``
+is given, all other size-related keyword arguments will be ignored. If ``size``
+is not provided, the function will attempt to infer the number of volumes from
+the length or value of the provided keyword argument. If more than one such
+keyword is provided, a :exc:`ValueError` will be raised.
 
-    """
-    uptake = kwargs.pop(UPTAKE_KWARG, None)
-    if uptake is None:
-        raise TypeError(KWARG_ERROR_MSG.format(kwarg=UPTAKE_KWARG))
-    return _value_iterator(
-        uptake,
-        ascending=False,
-        round_decimals=kwargs.pop("round_decimals", DEFAULT_ROUND_DECIMALS),
-    )
+Yields
+------
+:obj:`int`
+    The next index.
+
+Examples
+--------
+>>> list(monotonic_value_iterator(bvals=[0.0, 0.0, 1000.0, 1000.0, 700.0, 700.0, 2000.0, 2000.0, 0.0]))
+[0, 1, 8, 4, 5, 2, 3, 6, 7]
+>>> list(monotonic_value_iterator(uptake=[-1.23, 1.06, 1.02, 1.38, -1.46, -1.12, -1.19, 1.24, 1.05]))
+[3, 7, 1, 8, 2, 5, 6, 0, 4]
+"""
 
 
 def centralsym_iterator(**kwargs) -> Iterator[int]:

test/test_estimator.py

@@ -133,8 +133,8 @@ def mock_iterator(*_, **kwargs):
         ("random", iterators.random_iterator, "pet"),
         ("centralsym", iterators.centralsym_iterator, "dwi"),
         ("centralsym", iterators.centralsym_iterator, "pet"),
-        ("bvalue", iterators.bvalue_iterator, "dwi"),
-        ("uptake", iterators.uptake_iterator, "pet"),
+        ("monotonic_value", iterators.monotonic_value_iterator, "dwi"),
+        ("monotonic_value", iterators.monotonic_value_iterator, "pet"),
     ],
 )
 def test_estimator_iterator_index_match(
@@ -206,10 +206,13 @@ class DummyXForm:
         return
     elif strategy == "centralsym":
         expected_indices = list(iterator_func(size=n_vols))
-    elif strategy == "bvalue":
-        expected_indices = list(iterator_func(bvals=bvals))
-    elif strategy == "uptake":
-        expected_indices = list(iterator_func(uptake=uptake))
+    elif strategy == "monotonic_value":
+        if modality == "dwi":
+            expected_indices = list(iterator_func(bvals=bvals, ascending=True))
+        elif modality == "pet":
+            expected_indices = list(iterator_func(uptake=uptake, ascending=False))
+        else:
+            raise NotImplementedError(f"Modality {modality} not implemented")
     else:
         raise ValueError(f"Unknown strategy {strategy}")
 

test/test_iterators.py

@@ -31,11 +31,10 @@
     KWARG_ERROR_MSG,
     UPTAKE_KWARG,
     _value_iterator,
-    bvalue_iterator,
     centralsym_iterator,
     linear_iterator,
+    monotonic_value_iterator,
     random_iterator,
-    uptake_iterator,
 )
 
 
@@ -144,43 +143,46 @@ def test_centralsym_iterator(kwargs, expected):
     assert list(centralsym_iterator(**kwargs)) == expected
 
 
-def test_bvalue_iterator_error():
-    with pytest.raises(TypeError, match=KWARG_ERROR_MSG.format(kwarg=BVALS_KWARG)):
-        list(bvalue_iterator())
-
-
 @pytest.mark.parametrize(
-    "bvals, expected",
+    "kwargs",
     [
-        ([0, 700, 1200], [0, 1, 2]),
-        ([0, 0, 1000, 700], [0, 1, 3, 2]),
-        ([0, 1000, 1500, 700, 2000], [0, 3, 1, 2, 4]),
+        {},
+        {"bvals": None},
+        {"uptake": None},
+        {"bvals": None, "uptake": None},
     ],
 )
-def test_bvalue_iterator(bvals, expected):
-    obtained = list(bvalue_iterator(bvals=bvals))
-    assert set(obtained) == set(range(len(bvals)))
-    # Should be ordered by increasing bvalue
-    sorted_bvals = [bvals[i] for i in obtained]
-    assert sorted_bvals == sorted(bvals)
+def test_monotonic_value_iterator_error(kwargs):
+    with pytest.raises(
+        TypeError, match=KWARG_ERROR_MSG.format(kwarg=f"{BVALS_KWARG} or {UPTAKE_KWARG}")
+    ):
+        monotonic_value_iterator(**kwargs)
+
 
+def test_monotonic_value_iterator_sorting_preference():
+    result = list(monotonic_value_iterator(bvals=[700, 1000], uptake=[0.14, 0.23, 0.47]))
+    assert result == [0, 1]
 
-def test_uptake_iterator_error():
-    with pytest.raises(TypeError, match=KWARG_ERROR_MSG.format(kwarg=UPTAKE_KWARG)):
-        list(uptake_iterator())
+    result = list(monotonic_value_iterator(bvals=None, uptake=[0.14, 0.23, 0.47]))
+    assert result == [2, 1, 0]
 
 
 @pytest.mark.parametrize(
-    "uptake, expected",
+    "feature, values, expected",
     [
-        ([0.3, 0.2, 0.1], [0, 1, 2]),
-        ([0.2, 0.1, 0.3], [2, 1, 0]),
-        ([-1.02, 1.16, -0.56, 0.43], [1, 3, 2, 0]),
+        ("bvals", [0, 700, 1200], [0, 1, 2]),
+        ("bvals", [0, 0, 1000, 700], [0, 1, 3, 2]),
+        ("bvals", [0, 1000, 1500, 700, 2000], [0, 3, 1, 2, 4]),
+        ("uptake", [0.3, 0.2, 0.1], [0, 1, 2]),
+        ("uptake", [0.2, 0.1, 0.3], [2, 1, 0]),
+        ("uptake", [-1.02, 1.16, -0.56, 0.43], [1, 3, 2, 0]),
     ],
 )
-def test_uptake_iterator_valid(uptake, expected):
-    obtained = list(uptake_iterator(uptake=uptake))
-    assert set(obtained) == set(range(len(uptake)))
-    # Should be ordered by decreasing uptake
-    sorted_uptake = [uptake[i] for i in obtained]
-    assert sorted_uptake == sorted(uptake, reverse=True)
+def test_monotonic_value_iterator(feature, values, expected):
+    obtained = list(monotonic_value_iterator(**{feature: values}))
+    assert set(obtained) == set(range(len(values)))
+    # If b-values, should be ordered by increasing value; if uptake values,
+    # should be ordered by decreasing uptake
+    sorted_vals = [values[i] for i in obtained]
+    reverse = True if feature == "uptake" else False
+    assert sorted_vals == sorted(values, reverse=reverse)