add support for percent in taper and coord select

d-chambers · d-chambers · commit d75879eabf66 · 2025-08-30T20:44:36.000+01:00
diff --git a/dascore/compat.py b/dascore/compat.py
@@ -10,7 +10,7 @@
 from contextlib import suppress
 
 import numpy as np
-from numpy import floor, interp  # NOQA
+from numpy import floor, interp, ndarray  # NOQA
 from numpy.random import RandomState
 from rich.progress import Progress  # NOQA
 from scipy.interpolate import interp1d  # NOQA
diff --git a/dascore/core/coords.py b/dascore/core/coords.py
@@ -30,6 +30,7 @@
     get_factor_and_unit,
     get_quantity,
     get_quantity_str,
+    percent,
 )
 from dascore.utils.display import get_nice_text
 from dascore.utils.docs import compose_docstring
@@ -505,11 +506,26 @@ def simplify_units(self) -> Self:
         _, unit = get_factor_and_unit(self.units, simplify=True)
         return self.convert_units(unit)
 
-    def coord_range(self):
-        """Return a scaler value for the coordinate (e.g., number of seconds)."""
-        if not self.evenly_sampled:
+    def coord_range(self, exact: bool = True):
+        """
+        Return a scaler value for the coordinate range (e.g., number of seconds).
+
+        Parameters
+        ----------
+        exact
+            If true, only exact ranges are returned. This accounts for
+            spacing at the end of each sample. Consequently, exact is only
+            possible for evenly sampled coords. If false, just disregard
+            this if coord isnt't evenly sampled.
+
+        """
+        if not self.evenly_sampled and exact:
             raise CoordError("coord_range has to be called on an evenly sampled data.")
-        return self.max() - self.min() + self.step
+        step = getattr(self, "step", None)
+        coord_range = self.max() - self.min()
+        if step is not None:
+            coord_range += step
+        return coord_range
 
     @abc.abstractmethod
     def sort(self, reverse=False) -> tuple[BaseCoord, slice | ArrayLike]:
@@ -599,7 +615,12 @@ def _get_compatible_value(self, value, relative=False):
         """
         # strip units and v
         if hasattr(value, "units"):
-            value = convert_units(value.magnitude, self.units, value.units)
+            mag, unit = value.magnitude, value.units
+            if unit == percent:
+                value = (mag / 100.0) * self.coord_range(exact=False)
+                relative = True
+            else:
+                value = convert_units(value.magnitude, self.units, value.units)
         # if null or ... just return None
         if not is_array(value) and (pd.isnull(value) or value is Ellipsis):
             return None
diff --git a/dascore/proc/basic.py b/dascore/proc/basic.py
@@ -108,7 +108,7 @@ def _fast_attr_update(self, attrs):
     return self.__class__(self.data, **out)
 
 
-def equals(self: PatchType, other: Any, only_required_attrs=True) -> bool:
+def equals(self: PatchType, other: Any, only_required_attrs=True, close=False) -> bool:
     """
     Determine if the current patch equals another.
 
@@ -119,6 +119,9 @@ def equals(self: PatchType, other: Any, only_required_attrs=True) -> bool:
     only_required_attrs
         If True, only compare required attributes. This helps avoid issues
         with comparing histories or custom attrs of patches, for example.
+    close
+        If True, the data can be "close" using np.allclose, otherwise
+        all data must be equal.
     """
     # different types are not equal
     if not isinstance(other, type(self)):
@@ -145,8 +148,12 @@ def equals(self: PatchType, other: Any, only_required_attrs=True) -> bool:
         }
         if not_equal:
             return False
-
-    return np.equal(self.data, other.data).all()
+    # Test data equality or proximity.
+    if close and not np.allclose(self.data, other.data):
+        return False
+    elif not close and not np.equal(self.data, other.data).all():
+        return False
+    return True
 
 
 def update(
diff --git a/dascore/proc/taper.py b/dascore/proc/taper.py
@@ -7,31 +7,16 @@
 from operator import add
 
 import numpy as np
-from scipy.signal import windows  # the best operating system?
 
 from dascore.constants import PatchType
 from dascore.exceptions import ParameterError
 from dascore.units import Quantity
 from dascore.utils.docs import compose_docstring
 from dascore.utils.misc import broadcast_for_index
 from dascore.utils.patch import get_dim_axis_value, patch_function
+from dascore.utils.signal import WINDOW_FUNCTIONS, _get_window_function
 from dascore.utils.time import to_float
 
-TAPER_FUNCTIONS = dict(
-    barthann=windows.barthann,
-    bartlett=windows.bartlett,
-    blackman=windows.blackman,
-    blackmanharris=windows.blackmanharris,
-    bohman=windows.bohman,
-    hamming=windows.hamming,
-    hann=windows.hann,
-    cos=windows.hann,
-    nuttall=windows.nuttall,
-    parzen=windows.parzen,
-    triang=windows.triang,
-    ramp=windows.triang,
-)
-
 
 def _get_taper_slices(patch, kwargs):
     """Get slice for start/end of patch."""
@@ -42,7 +27,7 @@ def _get_taper_slices(patch, kwargs):
         start, stop = value[0], value[1]
     else:
         start, stop = value, value
-    dur = coord.max() - coord.min()
+    dur = coord.coord_range(exact=False)
     # either let units pass through or multiply by d_len
     clses = (Quantity, np.timedelta64)
     start = start if isinstance(start, clses) or start is None else start * dur
@@ -53,19 +38,6 @@ def _get_taper_slices(patch, kwargs):
     return axis, (start, stop), inds_1, inds_2
 
 
-def _get_window_function(window_type):
-    """Get the window function to use for taper."""
-    # get taper function or raise if it isn't known.
-    if window_type not in TAPER_FUNCTIONS:
-        msg = (
-            f"'{window_type}' is not a known window type. "
-            f"Options are: {sorted(TAPER_FUNCTIONS)}"
-        )
-        raise ParameterError(msg)
-    func = TAPER_FUNCTIONS[window_type]
-    return func
-
-
 def _validate_windows(samps, start_slice, end_slice, shape, axis):
     """Validate the windows don't overlap or exceed dim len."""
     max_len = shape[axis]
@@ -87,7 +59,7 @@ def _validate_windows(samps, start_slice, end_slice, shape, axis):
 
 
 @patch_function()
-@compose_docstring(taper_type=sorted(TAPER_FUNCTIONS))
+@compose_docstring(taper_type=sorted(WINDOW_FUNCTIONS))
 def taper(
     patch: PatchType,
     window_type: str = "hann",
@@ -219,7 +191,7 @@ def _get_range_envelope(coord, inds, window_type, invert):
 
 
 @patch_function()
-@compose_docstring(taper_type=sorted(TAPER_FUNCTIONS))
+@compose_docstring(taper_type=sorted(WINDOW_FUNCTIONS))
 def taper_range(
     patch: PatchType,
     window_type: str = "hann",
diff --git a/dascore/utils/signal.py b/dascore/utils/signal.py
@@ -0,0 +1,35 @@
+"""
+Utilities for signal processing.
+"""
+
+from scipy.signal import windows
+
+from dascore.exceptions import ParameterError
+
+WINDOW_FUNCTIONS = dict(
+    barthann=windows.barthann,
+    bartlett=windows.bartlett,
+    blackman=windows.blackman,
+    blackmanharris=windows.blackmanharris,
+    bohman=windows.bohman,
+    hamming=windows.hamming,
+    hann=windows.hann,
+    cos=windows.hann,
+    nuttall=windows.nuttall,
+    parzen=windows.parzen,
+    triang=windows.triang,
+    ramp=windows.triang,
+)
+
+
+def _get_window_function(window_type):
+    """Get the window function to use for taper."""
+    # get taper function or raise if it isn't known.
+    if window_type not in WINDOW_FUNCTIONS:
+        msg = (
+            f"'{window_type}' is not a known window type. "
+            f"Options are: {sorted(WINDOW_FUNCTIONS)}"
+        )
+        raise ParameterError(msg)
+    func = WINDOW_FUNCTIONS[window_type]
+    return func
diff --git a/tests/test_core/test_coords.py b/tests/test_core/test_coords.py
@@ -25,7 +25,7 @@
     get_coord,
 )
 from dascore.exceptions import CoordError, ParameterError
-from dascore.units import get_quantity
+from dascore.units import get_quantity, percent
 from dascore.utils.misc import all_close, register_func
 from dascore.utils.time import dtype_time_like, is_datetime64, is_timedelta64, to_float
 
@@ -357,9 +357,13 @@ def test_both_values_and_data_raises(self):
 
     def test_coord_range(self, monotonic_float_coord):
         """Ensure that coord_range raises an error for not evenly sampled patches."""
+        coord = monotonic_float_coord
         msg = "has to be called on an evenly sampled"
         with pytest.raises(CoordError, match=msg):
-            monotonic_float_coord.coord_range()
+            coord.coord_range()
+        # But when exact=False it should work.
+        out = coord.coord_range(exact=False)
+        assert out == (coord.max() - coord.min())
 
     def test_get_coord_datetime(self):
         """Ensure get_coord accepts a datetime object. See #467."""
@@ -745,6 +749,12 @@ def test_select_samples_no_int_raises(self, random_coord):
         with pytest.raises(ParameterError, match=expected):
             random_coord.select((1, 1.2), samples=True)
 
+    def test_percentage(self, coord):
+        """Ensure selecting by percentage works."""
+        out, indexer = coord.select((10 * percent, -20 * percent))
+        if coord.evenly_sampled:
+            assert abs((len(out) / len(coord)) - 0.70) < len(coord) / 100.0
+
 
 class TestOrder:
     """Tests for ordering coordinates."""
diff --git a/tests/test_core/test_patch.py b/tests/test_core/test_patch.py
@@ -401,6 +401,12 @@ def test_negative_equals(self, random_patch):
         assert -random_patch == -random_patch
         assert (-random_patch).abs() == random_patch
 
+    def test_close(self, random_patch):
+        """Test the `close` parameter"""
+        new = random_patch * 0.999999999999999
+        assert not new.equals(random_patch, close=False)
+        assert new.equals(random_patch, close=True)
+
 
 class TestTranspose:
     """Tests for switching dimensions."""
diff --git a/tests/test_proc/test_taper.py b/tests/test_proc/test_taper.py
@@ -7,9 +7,10 @@
 
 import dascore as dc
 from dascore.exceptions import ParameterError
-from dascore.proc.taper import TAPER_FUNCTIONS, taper
-from dascore.units import m
+from dascore.proc.taper import taper
+from dascore.units import m, percent
 from dascore.utils.misc import broadcast_for_index
+from dascore.utils.signal import WINDOW_FUNCTIONS
 
 gen = np.random.default_rng(32)
 
@@ -21,7 +22,7 @@ def patch_ones(random_patch):
     return patch
 
 
-@pytest.fixture(scope="session", params=sorted(TAPER_FUNCTIONS))
+@pytest.fixture(scope="session", params=sorted(WINDOW_FUNCTIONS))
 def time_tapered_patch(request, patch_ones):
     """Return a tapered trace."""
     # first get a patch with all ones for easy testing
@@ -39,6 +40,14 @@ def _get_start_end_indices(patch, dim):
     return inds_start, inds_end
 
 
+@pytest.fixture(scope="class")
+def patch_sorted_time(random_patch):
+    """Return a patch with sorted but not evenly spaced time dim."""
+    times = gen.random(len(random_patch.get_coord("time")))
+    new_times = dc.to_datetime64(np.sort(times))
+    return random_patch.update_coords(time=new_times)
+
+
 class TestTaperBasics:
     """Ensure each taper runs."""
 
@@ -125,17 +134,21 @@ def test_timedelta_taper(self, random_patch):
         patch2 = random_patch.taper(time=time2)
         assert patch1 == patch2
 
+    def test_percentage_taper(self, patch_ones):
+        """Ensure a percentage unit can be used in addition to fraction."""
+        out1 = patch_ones.taper(time=(0.1, 0.2))
+        out2 = patch_ones.taper(time=(10 * percent, 20 * percent))
+        assert out1.equals(out2, close=True)
+
+    def test_uneven_time_coord(self, patch_sorted_time):
+        """Ensure taper works on patches without even sampling."""
+        out = patch_sorted_time.taper(time=(0.1, 0.2))
+        assert isinstance(out, dc.Patch)
+
 
 class TestTaperRange:
     """Test for tapering a range of values."""
 
-    @pytest.fixture(scope="class")
-    def patch_sorted_time(self, random_patch):
-        """Return a patch with sorted but not evenly spaced time dim."""
-        times = gen.random(len(random_patch.get_coord("time")))
-        new_times = dc.to_datetime64(np.sort(times))
-        return random_patch.update_coords(time=new_times)
-
     def test_dims(self, patch_ones):
         """Ensure both dimensions work."""
         for ax, dim in enumerate(patch_ones.dims):
