Select, unselect, and split a spool along the fiber

dascore/core/patch.py

@@ -403,6 +403,7 @@ def get_patch_name(self, *args, **kwargs) -> str:
     # --- processing funcs
 
     select = dascore.proc.select
+    unselect = dascore.proc.unselect
     order = dascore.proc.order
 
     correlate = dascore.proc.correlate

dascore/io/index/planned.py

@@ -367,6 +367,8 @@ def __init__(
         mode: str = "chunk",
         check_behavior: WARN_LEVELS = "warn",
         origin_path=None,
+        stamped: tuple[str, ...] = (),
+        lossy: bool = False,
     ):
         if "output_id" not in member_rows.columns:
             msg = "member_rows must carry an output_id column."
@@ -382,6 +384,13 @@ def __init__(
         self.check_behavior = check_behavior
         # informational only: the directory/file the plan derived from
         self.origin_path = origin_path
+        # attrs the outputs state about themselves rather than inherit
+        self.stamped = tuple(stamped)
+        # Whether the outputs leave samples of their sources out. A lossy
+        # plan must never be collapsed: its members do not cover their
+        # sources, so re-planning over them would load back what it
+        # dropped. See `collapse_working_df`.
+        self.lossy = bool(lossy)
 
     def live_entries(self) -> dict[str, dc.Patch]:
         """Expose the loader's live registry (for absorption/transfer)."""
@@ -460,20 +469,37 @@ def resolve(self, row: Mapping, **trim) -> dc.Patch:
         if self.mode == "identity":
             # one untouched member per output; residuals apply at load
             assert len(members) == 1
-            return self._load_member(members.iloc[0].to_dict())
-        if self.mode == "concat":
-            patches = [
+            patch = self._load_member(members.iloc[0].to_dict())
+        elif self.mode == "concat":
+            loaded = [
                 self._load_member(kwargs) for kwargs in members.to_dict("records")
             ]
             out = concatenate_patches(
-                patches, check_behavior=self.check_behavior, **{self.dim: None}
+                loaded, check_behavior=self.check_behavior, **{self.dim: None}
             )
             assert len(out) == 1
-            return out[0]
-        joined = members.assign(current_index=output_id)
-        patches = self._assembler()._patch_from_instruction_df(joined)
-        assert len(patches) == 1
-        return patches[0]
+            patch = out[0]
+        else:
+            joined = members.assign(current_index=output_id)
+            assembled = self._assembler()._patch_from_instruction_df(joined)
+            assert len(assembled) == 1
+            patch = assembled[0]
+        return self._stamp(patch, row)
+
+    def _stamp(self, patch: dc.Patch, row: Mapping) -> dc.Patch:
+        """
+        Apply the attrs the outputs state about themselves, if any.
+
+        An output is assembled from its members, so it carries their
+        attrs and knows nothing of why it was cut out. `stamped` is how
+        an operation which does know says so -- `Spool.split_by`
+        recording which value each patch was split on -- and it keeps
+        the patch which comes out agreeing with the row `get_contents`
+        shows for it.
+        """
+        if not self.stamped:
+            return patch
+        return patch.update_attrs(**{x: row[x] for x in self.stamped})
 
 
 def _residual_ranges(residuals) -> dict:
@@ -506,6 +532,8 @@ def derived_catalog(
     mode: str = "chunk",
     check_behavior: WARN_LEVELS = "warn",
     origin_path=None,
+    stamped: tuple[str, ...] = (),
+    lossy: bool = False,
 ) -> PatchCatalog:
     """
     Materialize a plan into a fresh in-memory catalog.
@@ -574,6 +602,8 @@ def derived_catalog(
         mode=mode,
         check_behavior=check_behavior,
         origin_path=origin_path,
+        stamped=stamped,
+        lossy=lossy,
     )
     backend = get_backend(":memory:")
     coord_dims_map = {} if parent is None else parent.backend.coord_dims_map()
@@ -607,9 +637,16 @@ def collapse_working_df(catalog: PatchCatalog) -> pd.DataFrame | None:
     applied to the envelopes. (Planning a different dimension must keep
     the assembled boundaries, so its caller plans over the output rows
     instead and never collapses.)
+
+    A *lossy* plan is the exception and never collapses. Collapsing is
+    sound because the members of a chunk or a subdivision together cover
+    their sources, so a re-plan which merges them back is entitled to
+    load a source whole. A plan which drops samples — channel selection
+    keeping some channels of a patch and not others — breaks exactly
+    that, and collapsing it would quietly load back what it removed.
     """
     resolver = catalog.resolver
-    if not isinstance(resolver, PlanResolver):
+    if not isinstance(resolver, PlanResolver) or resolver.lossy:
         return None
     members = resolver.member_rows
     if catalog.is_view:

dascore/io/index/query.py

@@ -14,7 +14,7 @@
 import operator
 import re
 from collections.abc import Sequence
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
 from enum import Enum, auto
 
 import numpy as np
@@ -435,10 +435,17 @@ def evaluate_attr_predicate(values, name: str, value, units=None) -> np.ndarray:
     units = units or {}
 
     def selector(item):
-        """Type one selector value the way the index would."""
+        """
+        Type one selector value the way the index would, in its units.
+
+        The converted value is what the comparison needs, not just the
+        check that it converts: a stored coordinate in degrees against a
+        selector pint bases in radians would otherwise match nothing,
+        where the range form — which keeps its converted bounds — does.
+        """
         out = _coerce_scalar(item, kinds)
-        _to_target_unit(out, units.get(out.kind), name)
-        return out
+        value = _to_target_unit(out, units.get(out.kind), name)
+        return replace(out, value=value)
 
     def each(func) -> np.ndarray:
         return np.array([x is not None and bool(func(x)) for x in typed], dtype=bool)

dascore/proc/coords.py

@@ -442,6 +442,17 @@ def coords_from_df(
     return out
 
 
+def _check_coord_names(patch: PatchType, kwargs) -> None:
+    """Refuse a name the patch has no coordinate for, naming what it has."""
+    if not (invalid := set(kwargs) - set(patch.coords.coord_map)):
+        return
+    valid_list = sorted(patch.coords.coord_map)
+    msg = (
+        f"Coordinate(s) {sorted(invalid)} not found in patch coordinates: {valid_list}"
+    )
+    raise PatchCoordinateError(msg)
+
+
 @patch_function(history=None)
 @compose_docstring(select_params=select_values_description)
 def select(
@@ -543,15 +554,7 @@ def select(
       1999
 
     """
-    # Check for and raise on invalid kwargs.
-    if invalid_coords := set(kwargs) - set(patch.coords.coord_map):
-        invalid_list = sorted(invalid_coords)
-        valid_list = sorted(patch.coords.coord_map)
-        msg = (
-            f"Coordinate(s) {invalid_list} not found in patch coordinates: {valid_list}"
-        )
-        raise PatchCoordinateError(msg)
-
+    _check_coord_names(patch, kwargs)
     new_coords, data = patch.coords.select(
         **kwargs,
         array=patch.data,
@@ -566,6 +569,92 @@ def select(
     return patch.new(data=data, coords=new_coords)
 
 
+@patch_function(history=None)
+def unselect(
+    patch: PatchType, *, copy=False, relative=False, samples=False, **kwargs
+) -> PatchType:
+    """
+    Return the patch outside a selection.
+
+    The complement of [`Patch.select`](`dascore.Patch.select`): it takes
+    the same selectors and removes the samples that selection would have
+    kept. With one coordinate named that is exactly the complement; with
+    several, each is complemented on its own — see the note below.
+
+    Parameters
+    ----------
+    patch
+        The patch object.
+    copy
+        If True, copy the resulting data. This is needed so the old
+        array can get gc'ed and memory freed.
+    relative
+        If True, unselect ranges are relative to the start of coordinate, if
+        positive, or the end of the coordinate, if negative.
+    samples
+        If True, the query meaning is in samples.
+    **kwargs
+        Used to specify the coordinate on which data are unselected.
+
+    Examples
+    --------
+    >>> import dascore as dc
+    >>> from dascore.examples import get_example_patch
+    >>> patch = get_example_patch()
+    >>>
+    >>> # Drop meters 50 to 300, keeping what lies outside them.
+    >>> outside = patch.unselect(distance=(50, 300))
+    >>>
+    >>> # Drop the first ten distance samples.
+    >>> trimmed = patch.unselect(distance=(..., 10), samples=True)
+
+    Notes
+    -----
+    - Removing a range from the middle of a coordinate leaves a hole in
+      it, so the result is no longer evenly sampled and the coordinate
+      becomes a monotonic array. That is exactly what
+      [`Spool.unselect`](`dascore.core.spool.Spool.unselect`) refuses the
+      patches' *own* coordinates for: at spool level the complement of a
+      range is a hole in every patch rather than a choice between
+      patches. The coordinates an attached inventory defines along the
+      fiber it does accept, since removing one of those chooses which
+      channels a patch holds.
+
+    - Each named coordinate is complemented on its own. Selecting on two
+      coordinates keeps the samples in both ranges, and everything
+      outside that is a frame around them rather than a block, which no
+      array can hold — so `unselect` removes each named range instead,
+      which is the part of the complement that is expressible. Two
+      coordinates along one dimension therefore both take their range
+      out of it, leaving what neither removed.
+    """
+    _check_coord_names(patch, kwargs)
+    keep: dict[str, np.ndarray] = {}
+    for name, value in kwargs.items():
+        coord = patch.coords.coord_map[name]
+        dims = patch.coords.dim_map[name]
+        if len(dims) != 1:
+            msg = (
+                f"Coordinate {name!r} spans {list(dims)}, so removing a range "
+                "of it does not name samples of one dimension to drop."
+            )
+            raise PatchCoordinateError(msg)
+        # Asking select itself which samples it would keep is what stops
+        # the two from drifting: one selector cannot come to mean
+        # different things in select and its complement.
+        _, indexer = coord.select(value, relative=relative, samples=samples)
+        selected = np.zeros(len(coord), dtype=bool)
+        selected[indexer] = True
+        keep[dims[0]] = keep.get(dims[0], True) & ~selected
+    # Kept as sample numbers along each dimension rather than as a mask
+    # per coordinate: coordinates sharing a dimension are applied in
+    # separate passes, so the second mask would meet an already trimmed
+    # axis, and a dimension carrying no values of its own takes samples
+    # where it would refuse an array.
+    trims = {dim: np.flatnonzero(mask) for dim, mask in keep.items()}
+    return patch.select(**trims, samples=True, copy=copy)
+
+
 @patch_function(history=None)
 def order(
     patch: PatchType, *, copy=False, relative=False, samples=False, **kwargs