From c9f63fd6801436b2a1c1ce8fd18a5872b69fb1a6 Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Wed, 12 Aug 2026 18:17:37 +0200 Subject: [PATCH 1/3] Conform a spool to the inventory which describes it Add Spool.conform_to_inventory, the one eager step of the inventory workflow: resolve every row now, drop the patches the inventory does not describe, and subdivide a patch whose span crosses a change of optical path into one patch per epoch. Subdivision is a derived-catalog plan, not a dataframe rewrite, so the existing chunk machinery does the loading and len/get_contents stay exact. Each piece opens at the first sample at or after its boundary, which keeps the split faithful to half-open epochs and loses no sample to a boundary the sample grid does not share. A patch spanning a change of acquisition raises instead: its halves were recorded under two configurations, so no subdivision makes it one honest patch. resolve_contexts now sits on the same epoch walk, which fixes it refusing a row that crosses a bound nothing changes across -- something Patch.enrich has always allowed. --- dascore/core/spool.py | 251 ++++++++++++- dascore/proc/inventory.py | 140 +++++++- dascore/utils/chunk_plan.py | 93 +++++ tests/test_proc/test_proc_inventory.py | 480 +++++++++++++++++++++++++ 4 files changed, 931 insertions(+), 33 deletions(-) diff --git a/dascore/core/spool.py b/dascore/core/spool.py index 6c30d9cf..d35fedab 100644 --- a/dascore/core/spool.py +++ b/dascore/core/spool.py @@ -38,6 +38,7 @@ InvalidSpoolQueryError, MissingPatchError, ParameterError, + PatchError, UnresolvedPatchError, ) from dascore.utils.chunk_plan import ( @@ -46,6 +47,7 @@ _combined_dtype, _ensure_patch_id, build_chunk_plan, + build_subdivision_plan, samples_adjusted_envelopes, ) from dascore.utils.display import get_dascore_text, get_nice_text @@ -88,15 +90,114 @@ def _copy_public_dataframe(frame: pd.DataFrame) -> pd.DataFrame: _VALID_ON_UNRESOLVED = ("warn", "raise", "ignore") +# Conform decides membership rather than metadata, so its third option is +# removal rather than silence -- and silence is what "drop" then means. +_VALID_ON_UNRESOLVED_CONFORM = ("raise", "warn", "drop") _UNRESOLVED_WARNING = ( "The attached inventory does not describe every patch in this spool, and " "those it does not describe were not enriched. Use on_unresolved='raise' " "to see which, 'ignore' to silence this, or remove them from the spool " - "once Spool.prune_to_inventory exists." + "with Spool.conform_to_inventory." ) +def _resolution_columns(frame: pd.DataFrame) -> list | None: + """ + Return the columns a relation resolves against an inventory with. + + Resolution needs an identity and the instants to resolve it at, so a + relation offering either is None here. A spool whose patches carry no + `acquisition_key` has no identity to offer, and one whose time axis + is not physical — lag times from a correlation, say — has no + instants, which is the same thing `Patch.enrich` refuses to guess at. + """ + columns = [frame.get(x) for x in ("acquisition_key", "time_min", "time_max")] + physical = all(column is not None for column in columns) and all( + np.issubdtype(column.dtype, np.datetime64) for column in columns[1:] + ) + return columns if physical else None + + +def _first_few(items, limit: int = 5) -> str: + """Name a few of the offending rows, and say how many went unnamed.""" + listed = ", ".join(str(x) for x in items[:limit]) + extra = len(items) - limit + return listed if extra <= 0 else f"{listed} (and {extra} more)" + + +def _report_unconformed(rows: pd.DataFrame, on_unresolved: str) -> None: + """ + Say what conforming is about to drop, as loudly as it was asked to. + + Naming the files is the whole value of the loud policies: an + inventory which covers an archive apart from a handful of patches is + reporting a gap in itself as often as a stray file. + """ + if on_unresolved == "drop": + return + paths = list(rows["source_path"]) + msg = ( + f"The inventory does not describe {len(paths)} patch(es) in this " + f"spool: {_first_few(paths)}. Pass on_unresolved='drop' to remove " + "them silently, or 'warn' to be told and carry on." + ) + if on_unresolved == "raise": + raise UnresolvedPatchError(msg) + warnings.warn(msg, UserWarning, stacklevel=3) + + +def _check_one_acquisition(source_rows: pd.DataFrame, epochs) -> None: + """ + Refuse the patches whose acquisition changes partway through. + + Subdividing cannot reconcile these the way it reconciles a change of + optical path: the pieces would say the same recording ran under two + configurations, which is a file that should not exist rather than + one to reconcile. So this is not something `on_unresolved` waves + through — the inventory describes the patch twice, not not at all. + """ + conflicted = [ + f"{path} at {row.conflict}" + for path, row in zip(source_rows["source_path"], epochs, strict=True) + if row.conflict is not None + ] + if not conflicted: + return + msg = ( + f"{len(conflicted)} patch(es) span a change of acquisition, which " + f"subdividing cannot reconcile: {_first_few(conflicted)}. Select the " + "side you want, or correct the inventory." + ) + raise PatchError(msg) + + +def _check_subdividable(source_rows: pd.DataFrame, rows: pd.DataFrame, cuts) -> None: + """ + Refuse a patch which must be split but states no sampling interval. + + The pieces are found on the patch's own sample grid, which its time + step is the only description of. Backing off to the raw boundary + would silently drop the sample either side of it, and losing a + sample is a worse answer than saying so — the caller asked for + metadata reconciliation, not for the data to be restructured. + """ + bad = [ + f"{path} at {row_cuts[0]}" + for path, step, row_cuts in zip( + source_rows["source_path"], rows["time_step"], cuts, strict=True + ) + if row_cuts and (pd.isnull(step) or not step) + ] + if not bad: + return + msg = ( + f"{len(bad)} patch(es) must be subdivided at an epoch boundary but " + f"state no time step to find their samples with: {_first_few(bad)}." + ) + raise PatchError(msg) + + def _unstated(values) -> np.ndarray: """ Return a mask of the entries which state no value. @@ -994,12 +1095,6 @@ def _resolve_rows(self, ids) -> np.ndarray: """ Resolve each presented row to its inventory context, or to None. - Resolution needs an identity and the instants to resolve it at. A - spool whose patches carry no `acquisition_key` has no identity to - offer, and one whose time axis is not physical — lag times from a - correlation, say — has no instants, which is the same thing - `Patch.enrich` refuses to guess at. - This is where the relation is realized, which is why it is only reached for a name the index does not state for every row. """ @@ -1007,11 +1102,8 @@ def _resolve_rows(self, ids) -> np.ndarray: out = np.full(len(ids), None, dtype=object) df = self._df - columns = [df.get(x) for x in ("acquisition_key", "time_min", "time_max")] - physical = all(column is not None for column in columns) and all( - np.issubdtype(column.dtype, np.datetime64) for column in columns[1:] - ) - if not physical: + columns = _resolution_columns(df) + if columns is None: return out # Aligned by id rather than by position: the relation is realized # by a route of its own and need not present every row the id list @@ -1073,9 +1165,11 @@ def attach_inventory(self, inventory) -> Self: Notes ----- - This is the metadata half of the workflow: resolving the index - against the inventory, subdividing it at epoch boundaries, and - selecting on inventory tracks are not implemented yet. + Attaching promises nothing about the spool matching the + inventory; + [`conform_to_inventory`](`dascore.core.spool.Spool.conform_to_inventory`) + is what makes it so. Selecting on the coordinates an inventory + defines along the fiber is not implemented yet. """ if not isinstance(inventory, Inventory): msg = f"attach_inventory needs an Inventory, got {type(inventory)}." @@ -1135,8 +1229,10 @@ def enrich( Enriching never removes a patch: one the inventory does not describe comes out unchanged rather than missing, so an inventory covering part of an archive needs no pruning first. Deciding - membership will be `prune_to_inventory`'s job, and leaving it there is - what keeps this lazy — nothing resolves until a patch is pulled. + membership is + [`conform_to_inventory`](`dascore.core.spool.Spool.conform_to_inventory`)'s + job, and leaving it there is what keeps this lazy — nothing + resolves until a patch is pulled. Parameters ---------- @@ -1209,6 +1305,125 @@ def enrich( new._on_unresolved = on_unresolved return new + def conform_to_inventory( + self, + inventory=None, + *, + on_unresolved: Literal["raise", "warn", "drop"] = "raise", + ) -> Self: + """ + Return a spool the inventory describes exactly, patch for patch. + + The one eager step of the inventory workflow: every row is + resolved now, patches the inventory does not describe are + dropped, and a patch whose span crosses a change of optical path + is subdivided into one patch per epoch — so the spool can grow as + well as shrink. It is metadata work; no patch data is read. + + Subdivision is exact. Each piece begins at the first sample at or + after its epoch's boundary, so together they hold every sample + the patch held and hold none of them twice, and `len` and + `get_contents` describe the pieces rather than the original. + + Parameters + ---------- + inventory + The inventory to conform to. Defaults to the spool's attached + inventory; given one, it is attached as well. + on_unresolved + What to do with a patch the inventory does not describe — one + carrying no `acquisition_key`, one carrying a key the + inventory does not resolve to exactly one entry, or one + reaching outside every matching epoch. "raise" (the default) + fails and names them, "warn" drops them and says so, and + "drop" discards them silently, which is what an inventory + deliberately covering part of an archive wants. + + Raises + ------ + PatchError + If a patch spans a change of *acquisition*. Its two halves + were recorded under different configurations, so no + subdivision makes it one honest patch, and `on_unresolved` + does not cover it: the inventory describes such a patch + twice rather than not at all. + + Examples + -------- + >>> import dascore as dc + >>> from dascore.examples import inventory_patch_pair + >>> + >>> patch, inventory = inventory_patch_pair() + >>> spool = dc.spool(patch).attach_inventory(inventory) + >>> assert len(spool.conform_to_inventory()) == 1 + >>> + >>> # A patch the inventory says nothing about can be dropped. + >>> other = patch.update_attrs(acquisition_key="DAS.R2D1..OTHER") + >>> mixed = dc.spool([patch, other]).attach_inventory(inventory) + >>> assert len(mixed.conform_to_inventory(on_unresolved="drop")) == 1 + """ + from dascore.io.index.planned import derived_catalog # noqa: PLC0415 + from dascore.proc.inventory import ( # noqa: PLC0415 + _NO_EPOCHS, + resolve_row_epochs, + ) + + if on_unresolved not in _VALID_ON_UNRESOLVED_CONFORM: + msg = ( + f"on_unresolved must be one of {_VALID_ON_UNRESOLVED_CONFORM}, " + f"got {on_unresolved!r}." + ) + raise ParameterError(msg) + if inventory is None and self._inventory is None: + msg = ( + "Spool.conform_to_inventory needs an inventory: pass one, or " + "attach one first with Spool.attach_inventory." + ) + raise ParameterError(msg) + new = ( + self.__class__(self) + if inventory is None + else self.attach_inventory(inventory) + ) + source_rows, working = new._plan_frames() + # The two frames are one relation split by column, so a row of + # either is the same patch as the row beside it; the messages + # below name files from one while judging the other. + assert (source_rows["_patch_id"].to_numpy() == working["_patch_id"]).all() + columns = _resolution_columns(working) + epochs = ( + [_NO_EPOCHS] * len(working) + if columns is None + else resolve_row_epochs(new._inventory, *columns) + ) + _check_one_acquisition(source_rows, epochs) + described = np.array([x.described for x in epochs], dtype=bool) + if not described.all(): + _report_unconformed(source_rows[~described], on_unresolved) + kept = working[described].reset_index(drop=True) + cuts = [x.cuts for x, keep in zip(epochs, described, strict=True) if keep] + if not any(cuts): # nothing to subdivide: a filter is the whole job + return new._restrict_to_rows(kept["_patch_id"].to_numpy()) + sources = source_rows[described].reset_index(drop=True) + _check_subdividable(sources, kept, cuts) + catalog = derived_catalog( + source_rows=sources, + plan=build_subdivision_plan(kept, cuts, "time"), + parent=new._catalog, + merge_kwargs={}, + mode="chunk", + origin_path=new.spool_path, + ) + return new._new_from_catalog(catalog) + + def _restrict_to_rows(self, patch_ids) -> Self: + """Return the view holding only the named rows, in the same order.""" + ids = np.asarray(self._catalog._ordered_ids(), dtype=np.int64) + keep = np.isin(ids, np.asarray(patch_ids, dtype=np.int64)) + if keep.all(): + return self + return self._new_from_catalog(self._catalog.restrict(keep, ids=ids)) + def _enrichment(self): """Return how this spool enriches, or None if it does not.""" if self._inventory is None or self._enrich_kwargs is None: @@ -1224,7 +1439,7 @@ def _maybe_enrich(self, patch): return patch.enrich(self._inventory, **kwargs) except UnresolvedPatchError: # The inventory does not describe this patch. Dropping it is - # prune_to_inventory's job, so it comes out as it went in. + # conform_to_inventory's job, so it comes out as it went in. if on_unresolved == "raise": raise if on_unresolved == "warn" and not self._warned_unresolved: diff --git a/dascore/proc/inventory.py b/dascore/proc/inventory.py index 997dc865..d01ee5be 100644 --- a/dascore/proc/inventory.py +++ b/dascore/proc/inventory.py @@ -3,7 +3,7 @@ from __future__ import annotations from collections.abc import Sequence, Sized -from typing import Any, Literal, get_args +from typing import Any, Literal, NamedTuple, get_args import numpy as np import pandas as pd @@ -203,12 +203,14 @@ def _epoch_bounds(inventory, acquisition_key: Sequence[str]) -> np.ndarray: def resolve_contexts(inventory, keys, starts, ends) -> np.ndarray: """ - Resolve index rows to their inventory contexts, one resolution per epoch. + Resolve index rows to the one inventory context each has, or to None. Each row is the half-open span of one patch. A row the inventory does - not describe, and a row whose span crosses an epoch boundary — which - the inventory describes twice rather than not at all, and which - `conform_to_inventory` exists to subdivide — resolve to None. + not describe resolves to None, and so does one whose span crosses a + change of acquisition or optical path — the inventory describes that + row twice rather than not at all, and `conform_to_inventory` exists + to subdivide it. A row crossing an epoch bound which changes neither + still has one context, and gets it. Parameters ---------- @@ -223,6 +225,72 @@ def resolve_contexts(inventory, keys, starts, ends) -> np.ndarray: ------- An object array of `ResolvedContext` or None, one per row. """ + epochs = resolve_row_epochs(inventory, keys, starts, ends) + out = np.full(len(epochs), None, dtype=object) + for row, epoch in enumerate(epochs): + if epoch.described and not epoch.cuts and epoch.conflict is None: + out[row] = epoch.context + return out + + +class RowEpochs(NamedTuple): + """ + How one index row sits against the epochs of its acquisition_key. + + Attributes + ---------- + cuts + The instants inside the row at which its optical path changes; + empty for a row which stays within one epoch of everything. + described + Whether the inventory resolves the row, over its whole span. + conflict + The instant at which the row's *acquisition* changes, or None. + Subdividing cannot rescue this: the pieces would describe one + patch recorded under two configurations, which is a file that + should not exist rather than one to reconcile. + context + What the row resolves to where it begins, or None where it is + undescribed. A row with no cuts and no conflict resolves to this + over its whole span. + """ + + cuts: tuple + described: bool + conflict: Any + context: ResolvedContext | None + + +# What a row whose key names no entry at all knows about its epochs. +_NO_EPOCHS = RowEpochs((), False, None, None) + + +def resolve_row_epochs(inventory, keys, starts, ends) -> list[RowEpochs]: + """ + Report how each index row sits against the inventory's epochs. + + Where [`resolve_contexts`](`dascore.proc.inventory.resolve_contexts`) + asks for the one context a row has and gives up on a row with two, + this reports *where* the row's answers change, so a caller can split + it into pieces which each have one. A row the inventory does not + describe over its whole span is reported undescribed rather than + partly described: the piece it does describe is not what the caller + asked about, and keeping the edge simple is worth more than + salvaging it. + + Parameters + ---------- + inventory + The inventory to resolve against. + keys + Each row's acquisition_key. + starts, ends + Each row's first and last instant. + + Returns + ------- + A list of `RowEpochs`, one per row. + """ frame = pd.DataFrame( { "key": np.asarray(keys, dtype=object), @@ -230,7 +298,7 @@ def resolve_contexts(inventory, keys, starts, ends) -> np.ndarray: "end": np.asarray(ends, dtype="datetime64[ns]"), } ) - out = np.full(len(frame), None, dtype=object) + out = [_NO_EPOCHS] * len(frame) for key, sub in frame.groupby("key", sort=False): # The empty string is how a patch spells "no identity"; it names # no entry, which is not the same as naming a missing one. A @@ -240,24 +308,66 @@ def resolve_contexts(inventory, keys, starts, ends) -> np.ndarray: continue bounds = _epoch_bounds(inventory, codes) first, last = sub["start"].to_numpy(), sub["end"].to_numpy() + # Epoch i runs from bounds[i - 1] up to bounds[i], so a row spans + # the epochs from its start's index through its end's, and every + # epoch after the first opens at the bound which begins it — an + # instant which resolves that epoch for every row reaching it. starts_at = np.searchsorted(bounds, first, side="right") ends_at = np.searchsorted(bounds, last, side="right") # A row with no instant of its own says nothing about which epoch # applies, and resolving at NaT holds every epoch effective; that # is the whole inventory answering rather than one entry. - unresolved = (starts_at != ends_at) | np.isnat(first) | np.isnat(last) + undated = np.isnat(first) | np.isnat(last) contexts: dict[int, ResolvedContext | None] = {} - for epoch in np.unique(starts_at[~unresolved]): - when = first[(starts_at == epoch) & ~unresolved][0] - try: - contexts[int(epoch)] = inventory.resolve(key, time=when) - except InvalidInventoryError: - contexts[int(epoch)] = None - for row, epoch, skip in zip(sub.index, starts_at, unresolved, strict=True): - out[row] = None if skip else contexts[int(epoch)] + for position, (lo, hi) in enumerate(zip(starts_at, ends_at)): + if undated[position]: + continue + for epoch in range(int(lo), int(hi) + 1): + if epoch not in contexts: + when = bounds[epoch - 1] if epoch else first[position] + contexts[epoch] = _try_resolve(inventory, key, when) + for row, lo, hi, bad in zip( + sub.index, starts_at, ends_at, undated, strict=True + ): + if bad: + continue + resolved = [contexts[epoch] for epoch in range(int(lo), int(hi) + 1)] + if any(x is None for x in resolved): + continue + out[row] = _epoch_changes(resolved, bounds[int(lo) : int(hi)]) return out +def _try_resolve(inventory, key, when) -> ResolvedContext | None: + """Resolve one instant, or None where the inventory has no one entry.""" + try: + return inventory.resolve(key, time=when) + except InvalidInventoryError: + return None + + +def _epoch_changes(resolved: list, boundaries) -> RowEpochs: + """ + Reduce a row's consecutive contexts to what changes between them. + + Compared by identity: one inventory hands out the same objects for + the same epoch, so identity says exactly "the answer changed" — and + says it without dumping two model trees to find out. A bound the + answer does not change across is not a boundary this row crosses. + """ + cuts = [] + for previous, current, boundary in zip(resolved, resolved[1:], boundaries): + if ( + previous.network is not current.network + or previous.fiber_array is not current.fiber_array + or previous.acquisition is not current.acquisition + ): + return RowEpochs(tuple(cuts), True, boundary, resolved[0]) + if previous.optical_path is not current.optical_path: + cuts.append(boundary) + return RowEpochs(tuple(cuts), True, None, resolved[0]) + + def get_attr_values(inventory, contexts, name: str) -> list: """ Return the value each resolved context states for one attr name. diff --git a/dascore/utils/chunk_plan.py b/dascore/utils/chunk_plan.py index 2f7bcf32..cb73ae7c 100644 --- a/dascore/utils/chunk_plan.py +++ b/dascore/utils/chunk_plan.py @@ -1030,3 +1030,96 @@ def build_chunk_plan( ignore_index=True, ) return ChunkPlan(outputs, members, name, value, params) + + +def _snapped_cuts(cuts, start, step) -> list: + """ + Return each cut moved up to the first sample at or after it. + + A cut is an arbitrary value — an epoch boundary owes the sample grid + nothing — while the pieces are described by inclusive envelopes, + which can only name samples. Snapping *up* keeps the split faithful + to a half-open ``[start, end)`` interval: the sample a cut lands + exactly on opens the piece that cut opens, and a cut falling between + two samples leaves the earlier one behind. + """ + span = to_float(step) + out = [] + for cut in cuts: + index = int(np.ceil(to_float(cut - start) / span)) + # Cuts fall strictly inside the row (see the docstring's + # contract), so each one really does open a piece. + assert index >= 1 + # Two cuts inside one sample interval name one split: the epoch + # between them covers no sample of this row. + if (value := start + index * step) not in out: + out.append(value) + return out + + +def build_subdivision_plan(df: pd.DataFrame, cuts, name: str) -> ChunkPlan: + """ + Build a plan splitting each row of a relation at its own cut values. + + Unlike a chunk plan, no row ever meets another: each output is one + contiguous piece of exactly one source row, so the pieces of a row + partition its samples — none is dropped, none is duplicated, and a + row with no cuts passes through as a single unmodified output. This + is what an operation which must re-describe patches rather than + restructure them (`Spool.conform_to_inventory`) needs. + + Parameters + ---------- + df + The relation to subdivide; one row per patch. + cuts + One sequence of cut values per row, in the row's own units, each + strictly inside that row's envelope. A cut opens a new piece at + the first sample at or after it, so a row with `n` distinct cuts + becomes at most `n + 1` outputs. + name + The dimension being subdivided. + + Notes + ----- + Every cut row needs a usable step to find its sample grid with, so + callers must reject a null or zero step themselves — they are the + ones which can name the file it came from. + """ + min_name, max_name = f"{name}_min", f"{name}_max" + step_name = f"{name}_step" + assert {min_name, max_name, step_name}.issubset(df.columns) + df = _ensure_patch_id(df).reset_index(drop=True) + positions, lows, highs, modified = [], [], [], [] + for position, row_cuts in enumerate(cuts): + start, stop = df.at[position, min_name], df.at[position, max_name] + # Envelopes are value-ordered whatever the coordinate's + # orientation, so the grid is walked by the step's magnitude. + step = abs(df.at[position, step_name]) + assert not len(row_cuts) or (not pd.isnull(step) and step) + grid = _snapped_cuts(row_cuts, start, step) + bounds = [start, *grid] + for index, low in enumerate(bounds): + high = bounds[index + 1] - step if index + 1 < len(bounds) else stop + positions.append(position) + lows.append(low) + highs.append(high) + modified.append(bool(grid)) + ids = np.arange(len(positions), dtype=np.int64) + # Outputs are not file rows: source bookkeeping stays on the members, + # and the dimension's structural identity described the whole row. + outputs = df.iloc[positions].drop( + columns=["_patch_id", f"_{name}_def_key", *_SOURCE_COLUMNS], errors="ignore" + ) + outputs = outputs.assign(**{min_name: lows, max_name: highs, "output_id": ids}) + members = pd.DataFrame( + { + "output_id": ids, + "_patch_id": df["_patch_id"].to_numpy()[positions], + min_name: lows, + max_name: highs, + step_name: df[step_name].to_numpy()[positions], + "_modified": modified, + } + ) + return ChunkPlan(outputs.reset_index(drop=True), members, name, None, {}) diff --git a/tests/test_proc/test_proc_inventory.py b/tests/test_proc/test_proc_inventory.py index dc007faf..71298bf9 100644 --- a/tests/test_proc/test_proc_inventory.py +++ b/tests/test_proc/test_proc_inventory.py @@ -10,6 +10,7 @@ from collections.abc import Mapping import numpy as np +import pandas as pd import pytest from pydantic import ValidationError @@ -35,6 +36,7 @@ ) from dascore.examples import inventory_patch_pair from dascore.exceptions import ( + CoordMergeError, InvalidInventoryError, InvalidSpoolError, InvalidSpoolQueryError, @@ -1916,3 +1918,481 @@ def build(*paths): assert "optical_components.loss_db" not in build(many).get_names().coords assert "optical_components.loss_db" in build(many, one).get_names().coords assert "optical_components.loss_db" in build(one).get_names().coords + + +def _split_epochs(inventory, when, *, acquisitions=False, second=None): + """ + Return the inventory with its one path (or acquisition) split at `when`. + + The two halves meet exactly at `when`, which epochs being half-open + makes the first instant of the second half rather than the last of + the first. + """ + array = inventory.networks[0].fiber_arrays[0] + old = array.acquisitions[0] if acquisitions else array.optical_paths[0] + changed = {} if second is None else second + halves = (old.new(end_time=when), old.new(start_time=when, **changed)) + field = "acquisitions" if acquisitions else "optical_paths" + return inventory.replace(array, array.new(**{field: halves})).check() + + +def _pieces(spool): + """The (min, max) time envelope of each patch a spool presents.""" + contents = spool.get_contents() + return list(zip(contents["time_min"], contents["time_max"])) + + +@pytest.fixture(scope="module") +def off_grid_boundary(patch): + """An epoch boundary a third of a sample past one of the patch's samples.""" + coord = patch.get_coord("time") + return coord.min() + (coord.max() - coord.min()) / 2 + coord.step / 3 + + +@pytest.fixture(scope="module") +def path_epochs(inventory, off_grid_boundary): + """The example inventory, its optical path split into two epochs.""" + return _split_epochs(inventory, off_grid_boundary, second={"name": "moved"}) + + +class TestConformBoundaryPolicy: + """Where a subdivided patch is cut, and which piece each sample joins.""" + + def test_split_is_lossless(self, patch, path_epochs): + """ + The pieces hold every sample the patch held, in order, once. + + Subdivision reconciles metadata; it must not restructure the + data, so a sample dropped or duplicated at the seam would be the + one failure the whole operation cannot afford. + """ + spool = dc.spool(patch).conform_to_inventory(path_epochs) + assert len(spool) == 2 + first, second = spool[0], spool[1] + times = np.concatenate( + [first.get_coord("time").values, second.get_coord("time").values] + ) + assert np.array_equal(times, patch.get_coord("time").values) + data = np.concatenate([first.data, second.data], axis=1) + assert np.array_equal(data, patch.data) + + def test_off_grid_boundary_keeps_its_straddling_sample( + self, patch, path_epochs, off_grid_boundary + ): + """ + The sample between `boundary - step` and `boundary` is not lost. + + An epoch boundary owes the sample grid nothing, so the naive + split of `[t_min, boundary - step]` and `[boundary, t_max]` + excludes any sample falling in between them from both pieces. + Cutting on the grid instead is what makes the split exact. + """ + coord = patch.get_coord("time") + naive = off_grid_boundary - coord.step + (_, first_end), (second_start, _) = _pieces( + dc.spool(patch).conform_to_inventory(path_epochs) + ) + # the sample the naive convention would have dropped + assert naive < first_end < off_grid_boundary + assert second_start == first_end + coord.step + + def test_boundary_sample_opens_the_second_piece(self, patch, inventory): + """ + A boundary landing exactly on a sample gives it to the new epoch. + + Epochs are half-open (`is_effective_at` admits `start` and + refuses `end`), so the split has to agree with them about which + epoch owns the instant they meet at. + """ + coord = patch.get_coord("time") + on_grid = coord.min() + coord.step * 10 + split = _split_epochs(inventory, on_grid, second={"name": "moved"}) + (_, first_end), (second_start, _) = _pieces( + dc.spool(patch).conform_to_inventory(split) + ) + assert second_start == on_grid + assert first_end == on_grid - coord.step + + def test_unstated_time_step_raises_naming_the_file(self, patch, inventory): + """ + A patch which must be cut but states no step says so, loudly. + + The cut is found on the patch's own sample grid, which the step + is the only description of. Backing off to the raw boundary + would silently drop the sample beside it, and for an operation + asked to reconcile metadata that is the worse answer. + """ + coord = patch.get_coord("time") + times = np.sort(np.concatenate([coord.values[:5], coord.values[10:15]])) + uneven = patch.select(time=(0, 10), samples=True).update_coords(time=times) + assert pd.isnull(dc.spool(uneven).get_contents()["time_step"]).all() + split = _split_epochs(inventory, times[3], second={"name": "moved"}) + with pytest.raises(PatchError, match="no time step"): + dc.spool(uneven).conform_to_inventory(split) + + def test_a_row_with_no_instants_is_undescribed(self, patch, path_epochs): + """ + A patch whose instants are unknown resolves to every epoch at once. + + `is_effective_at(NaT)` holds every epoch effective, so such a row + would be answered by the whole inventory rather than by one + entry. That is ambiguity rather than resolution, so + `on_unresolved` governs it — the same call `Patch.enrich` makes. + """ + undated = patch.update_coords( + time=np.full(patch.shape[1], np.datetime64("NaT")) + ).update_attrs(tag="undated") + # alongside a dated patch, so the row really does carry NaT + # instants rather than a time column of another kind entirely + spool = dc.spool([patch, undated]).attach_inventory(path_epochs) + assert pd.isnull(spool.get_contents()["time_min"]).any() + out = spool.conform_to_inventory(on_unresolved="drop") + assert set(out.get_contents()["tag"]) == {"random"} + with pytest.raises(UnresolvedPatchError, match="does not describe"): + spool.conform_to_inventory() + + def test_an_empty_key_is_undescribed(self, patch, path_epochs): + """ + The empty string is how a patch spells "no identity" at all. + + It names no entry, which is not the same as naming a missing + one, and neither is something to resolve. + """ + bare = patch.update_attrs(acquisition_key="", tag="bare") + spool = dc.spool([patch, bare]).attach_inventory(path_epochs) + out = spool.conform_to_inventory(on_unresolved="drop") + assert set(out.get_contents()["tag"]) == {"random"} + + def test_a_relative_time_axis_is_undescribed(self, patch, path_epochs): + """ + Lag times are not instants, so they pick no epoch either. + + Reading them as instants since 1970 would choose an epoch from + an offset; `Patch.enrich` refuses such a patch outright, and + conform has a policy for it instead of a guess. + """ + coord = patch.get_coord("time") + lags = patch.update_coords(time=coord.values - coord.min()) + spool = dc.spool(lags).attach_inventory(path_epochs) + assert len(spool.conform_to_inventory(on_unresolved="drop")) == 0 + with pytest.raises(UnresolvedPatchError, match="does not describe"): + spool.conform_to_inventory() + + def test_a_merged_patch_carries_one_key(self, patch, path_epochs): + """ + Conform needs one acquisition_key per row, and merging leaves it one. + + `acquisition_key` is a default group attribute, so patches from + two acquisitions partition apart and no output is ever built + from both. Grouping the partition away instead makes the key a + conflicted column, which merging refuses outright. + """ + other = patch.update_attrs(acquisition_key="DAS.R2D1..OTHER") + spool = dc.spool([patch, other]) + assert "acquisition_key" in dc.get_config().groupby_attrs + merged = spool.chunk(time=None) + assert sorted(merged.get_contents()["acquisition_key"]) == [ + "DAS.R2D1..OTHER", + "DAS.R2D1..RAW", + ] + conformed = merged.conform_to_inventory(path_epochs, on_unresolved="drop") + assert set(conformed.get_contents()["acquisition_key"]) == {"DAS.R2D1..RAW"} + with pytest.raises(CoordMergeError, match="acquisition_key"): + spool.chunk(time=None, group="tag") + + def test_a_merge_which_drops_the_key_is_undescribed(self, patch, path_epochs): + """ + A row whose merge discarded the key resolves to nothing, loudly. + + `conflict="drop"` is the one way an output can carry no identity + at all, and conform's own default reports it rather than + quietly returning an empty spool. + """ + other = patch.update_attrs(acquisition_key="DAS.R2D1..OTHER") + merged = dc.spool([patch, other]).chunk(time=None, group="tag", conflict="drop") + assert "acquisition_key" not in merged.get_contents().columns + with pytest.raises(UnresolvedPatchError, match="does not describe"): + merged.conform_to_inventory(path_epochs) + + +class TestConformMembership: + """Which patches a conformed spool holds, and how it says so.""" + + def test_a_described_spool_is_unchanged(self, patch, inventory): + """An inventory which already describes everything removes nothing.""" + spool = dc.spool(patch).attach_inventory(inventory) + assert len(spool.conform_to_inventory()) == len(spool) == 1 + + def test_undescribed_patches_drop(self, patch, inventory): + """A patch naming an entry the inventory lacks is not conformable.""" + other = patch.update_attrs(acquisition_key="DAS.R2D1..OTHER", tag="other") + spool = dc.spool([patch, other]).attach_inventory(inventory) + out = spool.conform_to_inventory(on_unresolved="drop") + assert out.get_contents()["tag"].tolist() == ["random"] + + def test_warn_drops_and_names_them(self, patch, inventory): + """The middle policy removes the rows but says which they were.""" + other = patch.update_attrs(acquisition_key="DAS.R2D1..OTHER", tag="other") + spool = dc.spool([patch, other]).attach_inventory(inventory) + with pytest.warns(UserWarning, match="does not describe"): + out = spool.conform_to_inventory(on_unresolved="warn") + assert len(out) == 1 + + def test_raise_is_the_default(self, patch, inventory): + """Conforming to an inventory which covers less says so by default.""" + other = patch.update_attrs(acquisition_key="DAS.R2D1..OTHER") + spool = dc.spool([patch, other]).attach_inventory(inventory) + with pytest.raises(UnresolvedPatchError, match="does not describe"): + spool.conform_to_inventory() + + def test_len_matches_the_contents(self, patch, path_epochs): + """ + Conforming is metadata work, so the count it reports is final. + + Both the subdivided and the merely-filtered route have to agree + with what iteration will actually yield. + """ + other = patch.update_attrs(acquisition_key="DAS.R2D1..OTHER") + spool = dc.spool([patch, other]).attach_inventory(path_epochs) + for out in ( + spool.conform_to_inventory(on_unresolved="drop"), + dc.spool(other).conform_to_inventory(path_epochs, on_unresolved="drop"), + ): + assert len(out) == len(out.get_contents()) == len(list(out)) + + def test_needs_an_inventory(self, patch): + """With none attached and none passed there is nothing to conform to.""" + with pytest.raises(ParameterError, match="needs an inventory"): + dc.spool(patch).conform_to_inventory() + + def test_an_empty_spool_conforms_to_nothing(self, inventory): + """No rows to resolve is not a reason to fail.""" + assert len(dc.spool([]).conform_to_inventory(inventory)) == 0 + + def test_policy_is_checked(self, patch, inventory): + """A misspelled policy is caught before any row is resolved.""" + spool = dc.spool(patch).attach_inventory(inventory) + with pytest.raises(ParameterError, match="on_unresolved must be"): + spool.conform_to_inventory(on_unresolved="ignore") + + def test_a_passed_inventory_attaches(self, patch, inventory): + """Passing one is also how to attach it, as with enrich.""" + out = dc.spool(patch).conform_to_inventory(inventory) + assert out._inventory is inventory + assert out.enrich()[0].attrs.gauge_length == 10.0 + + def test_patches_without_keys_are_undescribed(self, inventory): + """A spool whose rows name no entry has nothing to resolve with.""" + spool = dc.spool([dc.get_example_patch()]).attach_inventory(inventory) + assert "acquisition_key" not in spool.get_contents().columns + assert len(spool.conform_to_inventory(on_unresolved="drop")) == 0 + + +class TestConformSubdivision: + """Patches the inventory describes twice, split into pieces it describes once.""" + + def test_the_spool_grows(self, patch, path_epochs): + """One patch spanning two path epochs becomes two patches.""" + spool = dc.spool(patch).attach_inventory(path_epochs) + assert len(spool) == 1 + assert len(spool.conform_to_inventory()) == 2 + + def test_each_piece_enriches_from_its_own_epoch(self, patch, inventory): + """ + Subdividing is what makes enrichment possible at all here. + + The undivided patch is described twice, which `Patch.enrich` + refuses; each piece is described once and takes the geometry of + the epoch it falls in. + """ + coord = patch.get_coord("time") + when = coord.min() + (coord.max() - coord.min()) / 2 + moved = inventory.networks[0].fiber_arrays[0].optical_paths[0].annotations[0] + split = _split_epochs( + inventory, + when, + second={ + "annotations": ( + moved.new(value="moved", start_distance=100.0, end_distance=400.0), + ) + }, + ) + spool = dc.spool(patch).attach_inventory(split) + with pytest.raises(PatchError, match="spans a change"): + patch.enrich(split) + first, second = spool.conform_to_inventory().enrich(coords=True) + assert set(np.unique(first.get_coord("zone").values)) == {"north", "south"} + assert set(np.unique(second.get_coord("zone").values)) == {"moved"} + + def test_several_epochs_split_a_patch_several_ways(self, patch, inventory): + """A row is cut once per boundary it crosses, not once at all.""" + coord = patch.get_coord("time") + array = inventory.networks[0].fiber_arrays[0] + old = array.optical_paths[0] + edges = [coord.min() + coord.step * x for x in (500, 1000)] + paths = ( + old.new(end_time=edges[0]), + old.new(start_time=edges[0], end_time=edges[1], name="middle"), + old.new(start_time=edges[1], name="last"), + ) + split = inventory.replace(array, array.new(optical_paths=paths)).check() + out = dc.spool(patch).conform_to_inventory(split) + assert len(out) == 3 + starts = [x[0] for x in _pieces(out)] + assert starts[1] == edges[0] and starts[2] == edges[1] + + def test_an_epoch_shorter_than_one_sample_makes_one_cut(self, patch, inventory): + """ + Two boundaries inside one sample interval name one split. + + The epoch between them covers no sample of this patch, so a + piece for it would hold nothing; the samples still divide into + the epoch before and the epoch after. + """ + coord = patch.get_coord("time") + array = inventory.networks[0].fiber_arrays[0] + old = array.optical_paths[0] + edges = [coord.min() + coord.step * 500 + x for x in (-coord.step / 3, 0)] + paths = ( + old.new(end_time=edges[0]), + old.new(start_time=edges[0], end_time=edges[1], name="blink"), + old.new(start_time=edges[1], name="after"), + ) + split = inventory.replace(array, array.new(optical_paths=paths)).check() + out = dc.spool(patch).conform_to_inventory(split) + assert len(out) == 2 + assert _pieces(out)[1][0] == edges[1] + assert out[0].shape[1] + out[1].shape[1] == patch.shape[1] + + def test_a_boundary_outside_the_patch_cuts_nothing(self, patch, inventory): + """Only the epochs a patch actually reaches into can divide it.""" + coord = patch.get_coord("time") + split = _split_epochs( + inventory, coord.max() + coord.step * 10, second={"name": "later"} + ) + assert len(dc.spool(patch).conform_to_inventory(split)) == 1 + + def test_a_boundary_which_changes_nothing_cuts_nothing(self, patch, inventory): + """ + An epoch bound the answer does not change across is not a boundary. + + Every stated time on the key's branch is a candidate, but only + the ones which actually resolve to a different path are cuts; + splitting on the rest would grow the spool for nothing. + """ + coord = patch.get_coord("time") + array = inventory.networks[0].fiber_arrays[0] + # the network's own epoch opens before the patch and closes after + network = inventory.networks[0] + bounded = inventory.replace( + network, + network.new( + start_time=coord.min() - coord.step, + end_time=coord.max() + coord.step, + fiber_arrays=(array,), + ), + ).check() + assert len(dc.spool(patch).conform_to_inventory(bounded)) == 1 + + def test_selection_reads_across_a_bound_which_changes_nothing( + self, patch, inventory + ): + """ + Selecting resolves a row crossing a bound its answers survive. + + `Patch.enrich` has always allowed this — it refuses only a real + change of acquisition or path — so a spool refusing it would + have made the two disagree about the same patch. + """ + coord = patch.get_coord("time") + network = inventory.networks[0] + bounded = inventory.replace( + network, + network.new( + start_time=coord.min() - coord.step, + end_time=coord.max() + coord.step, + ), + ).check() + spool = dc.spool(patch).attach_inventory(bounded) + assert len(spool.select(gauge_length=10.0)) == 1 + assert patch.enrich(bounded).attrs.gauge_length == 10.0 + + def test_an_acquisition_change_raises(self, patch, inventory, off_grid_boundary): + """ + No subdivision makes a patch recorded two ways into one patch. + + The inventory describes it twice, so `on_unresolved` — which is + about patches it does not describe — has no say either. + """ + split = _split_epochs( + inventory, + off_grid_boundary, + acquisitions=True, + second={"gauge_length": 20.0}, + ) + spool = dc.spool(patch).attach_inventory(split) + for policy in ("raise", "warn", "drop"): + with pytest.raises(PatchError, match="span a change of acquisition"): + spool.conform_to_inventory(on_unresolved=policy) + + +class TestConformComposition: + """How conforming sits with the rest of the spool API.""" + + def test_conform_is_idempotent(self, patch, path_epochs): + """The pieces of a conformed spool each sit inside one epoch.""" + once = dc.spool(patch).conform_to_inventory(path_epochs) + twice = once.conform_to_inventory() + assert len(twice) == len(once) == 2 + assert _pieces(twice) == _pieces(once) + + def test_conform_nests_over_a_chunked_spool(self, patch, inventory): + """ + Conforming a chunked spool splits the chunks, not their sources. + + The pieces are cut out of the assembled outputs, so a chunk + which straddles a boundary becomes two and the rest stand. + """ + coord = patch.get_coord("time") + when = coord.min() + np.timedelta64(6, "s") + split = _split_epochs(inventory, when, second={"name": "moved"}) + chunked = dc.spool(patch).chunk(time=4) + assert len(chunked) == 2 + out = chunked.conform_to_inventory(split) + assert len(out) == 3 + assert [x[0] for x in _pieces(out)][2] == when + assert out[2].get_coord("time").min() == when + + def test_selecting_after_conforming(self, patch, path_epochs, off_grid_boundary): + """The pieces are ordinary rows, so ordinary selection reaches them.""" + out = dc.spool(patch).conform_to_inventory(path_epochs) + late = out.select(time=(off_grid_boundary, None)) + assert len(late) == 1 + assert late[0].get_coord("time").min() >= off_grid_boundary + + def test_the_inventory_carries_over(self, patch, path_epochs): + """A conformed spool still knows what it was conformed to.""" + out = dc.spool(patch).attach_inventory(path_epochs).conform_to_inventory() + assert out._inventory is path_epochs + assert len(out.select(gauge_length=10.0)) == 2 + + def test_conform_over_a_directory_spool(self, tmp_path_factory, patch, inventory): + """ + The pieces of a file-backed patch are read back exactly. + + In-memory patches ignore the trim hints a plan passes down, so + only a real file exercises the read path the pieces take. + """ + coord = patch.get_coord("time") + path = tmp_path_factory.mktemp("conform_directory") + dc.write(patch, path / "patch.h5", "dasdae") + split = _split_epochs( + inventory, coord.min() + coord.step * 500, second={"name": "moved"} + ) + out = dc.spool(path).update().conform_to_inventory(split) + assert len(out) == 2 + first, second = out[0], out[1] + assert first.shape[1] == 500 + assert second.shape[1] == patch.shape[1] - 500 + data = np.concatenate([first.data, second.data], axis=1) + assert np.array_equal(data, patch.data) From 919eb03af8197e80daed1082fcb34568c079ad18 Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Wed, 12 Aug 2026 18:44:34 +0200 Subject: [PATCH 2/3] Address the adversarial review of conform_to_inventory Snapping a cut onto the sample grid divided in float seconds, so an exactly-on-grid boundary could round a hair above its index and take the boundary sample with it into the epoch before the boundary -- the one place it must not go. Three reviewers found it independently, and the fudge factor CoordRange._get_index uses is not enough here: past a million samples the error outgrows any fixed tolerance. The ratio now only starts the search and the grid itself settles it, which is exact either way it errs. Comparing resolutions by identity made conform refuse patches enrich accepts: a fiber array re-registered with a new description resolves to a fresh object whose acquisition and optical path say exactly what they said before. Compare what the entries say, as _resolve_context does, and compare only the two which say anything about the patch. Also: a row whose end precedes its start no longer raises IndexError out of Spool.select; enrich and conform share their argument checking; and the two tests named for a boundary that changes nothing now put one inside the patch, where the comparison they pin actually runs. --- dascore/core/spool.py | 150 ++++++++++++--------- dascore/proc/inventory.py | 61 ++++++--- dascore/utils/chunk_plan.py | 27 +++- tests/test_proc/test_proc_inventory.py | 178 +++++++++++++++++++++---- tests/test_utils/test_chunk.py | 88 +++++++++++- 5 files changed, 385 insertions(+), 119 deletions(-) diff --git a/dascore/core/spool.py b/dascore/core/spool.py index d35fedab..827e692d 100644 --- a/dascore/core/spool.py +++ b/dascore/core/spool.py @@ -90,8 +90,9 @@ def _copy_public_dataframe(frame: pd.DataFrame) -> pd.DataFrame: _VALID_ON_UNRESOLVED = ("warn", "raise", "ignore") -# Conform decides membership rather than metadata, so its third option is -# removal rather than silence -- and silence is what "drop" then means. +# Conform decides membership, not metadata, so where enrich's quiet option +# is "ignore" -- leave the patch as it was -- conform's is "drop": every +# policy here removes the patch, and only the noise differs. _VALID_ON_UNRESOLVED_CONFORM = ("raise", "warn", "drop") _UNRESOLVED_WARNING = ( @@ -107,7 +108,7 @@ def _resolution_columns(frame: pd.DataFrame) -> list | None: Return the columns a relation resolves against an inventory with. Resolution needs an identity and the instants to resolve it at, so a - relation offering either is None here. A spool whose patches carry no + relation missing either gets None here. A spool whose patches carry no `acquisition_key` has no identity to offer, and one whose time axis is not physical — lag times from a correlation, say — has no instants, which is the same thing `Patch.enrich` refuses to guess at. @@ -137,10 +138,16 @@ def _report_unconformed(rows: pd.DataFrame, on_unresolved: str) -> None: if on_unresolved == "drop": return paths = list(rows["source_path"]) + advice = ( + "Pass on_unresolved='warn' to drop them with a warning, or 'drop' " + "to drop them silently." + if on_unresolved == "raise" + else "Pass on_unresolved='drop' to silence this, or 'raise' to fail " + "on the gap instead." + ) msg = ( f"The inventory does not describe {len(paths)} patch(es) in this " - f"spool: {_first_few(paths)}. Pass on_unresolved='drop' to remove " - "them silently, or 'warn' to be told and carry on." + f"spool: {_first_few(paths)}. {advice}" ) if on_unresolved == "raise": raise UnresolvedPatchError(msg) @@ -152,10 +159,9 @@ def _check_one_acquisition(source_rows: pd.DataFrame, epochs) -> None: Refuse the patches whose acquisition changes partway through. Subdividing cannot reconcile these the way it reconciles a change of - optical path: the pieces would say the same recording ran under two - configurations, which is a file that should not exist rather than - one to reconcile. So this is not something `on_unresolved` waves - through — the inventory describes the patch twice, not not at all. + optical path — see `RowEpochs.conflict` for why. `on_unresolved` does + not wave them through either: the inventory describes such a patch + twice rather than not at all. """ conflicted = [ f"{path} at {row.conflict}" @@ -177,10 +183,12 @@ def _check_subdividable(source_rows: pd.DataFrame, rows: pd.DataFrame, cuts) -> Refuse a patch which must be split but states no sampling interval. The pieces are found on the patch's own sample grid, which its time - step is the only description of. Backing off to the raw boundary - would silently drop the sample either side of it, and losing a - sample is a worse answer than saying so — the caller asked for - metadata reconciliation, not for the data to be restructured. + step is the only description of: a piece ends one step short of + where the next begins. Without a step both pieces would have to + claim the boundary instant, and envelopes are inclusive, so a sample + landing there would appear in both. Duplicating a sample is a worse + answer than saying so — the caller asked for metadata + reconciliation, not for the data to be restructured. """ bad = [ f"{path} at {row_cuts[0]}" @@ -977,9 +985,7 @@ def unselect( # The complement is taken against select itself rather than by # negating each predicate, so the two can never drift apart. removed = self.select(_attrs=stated)._catalog._ordered_ids() - ids = np.asarray(self._catalog._ordered_ids(), dtype=np.int64) - keep = ~np.isin(ids, np.asarray(removed, dtype=np.int64)) - return self._new_from_catalog(self._catalog.restrict(keep, ids=ids)) + return self._restrict_to_rows(removed, keep=False) def _index_names(self) -> tuple[set[str], set[str]]: """The attr and coord names the index knows, read once per call.""" @@ -1284,22 +1290,8 @@ def enrich( # Settled now rather than on extraction: a misspelled argument # should be an error here, not on some patch pulled much later. enrich_kwargs = _normalize_enrich_kwargs(kwargs) - if on_unresolved not in _VALID_ON_UNRESOLVED: - msg = ( - f"on_unresolved must be one of {_VALID_ON_UNRESOLVED}, " - f"got {on_unresolved!r}." - ) - raise ParameterError(msg) - if inventory is None and self._inventory is None: - msg = ( - "Spool.enrich needs an inventory: pass one, or attach one " - "first with Spool.attach_inventory." - ) - raise ParameterError(msg) - new = ( - self.__class__(self) - if inventory is None - else self.attach_inventory(inventory) + new = self._with_inventory( + inventory, on_unresolved, _VALID_ON_UNRESOLVED, "enrich" ) new._enrich_kwargs = enrich_kwargs new._on_unresolved = on_unresolved @@ -1317,24 +1309,32 @@ def conform_to_inventory( The one eager step of the inventory workflow: every row is resolved now, patches the inventory does not describe are dropped, and a patch whose span crosses a change of optical path - is subdivided into one patch per epoch — so the spool can grow as - well as shrink. It is metadata work; no patch data is read. + is subdivided at each such change — so the spool can grow as well + as shrink. A bound the answers survive unchanged is not a change, + and does not divide anything. It is metadata work; no patch data + is read. Subdivision is exact. Each piece begins at the first sample at or - after its epoch's boundary, so together they hold every sample - the patch held and hold none of them twice, and `len` and + after the change which opens it, so together they hold every + sample the patch held and hold none of them twice, and `len` and `get_contents` describe the pieces rather than the original. Parameters ---------- inventory The inventory to conform to. Defaults to the spool's attached - inventory; given one, it is attached as well. + inventory; given one, it is attached as well — and attaching + clears enrichment set up from the old one, as it does + everywhere. Conforming to the spool's own inventory leaves + enrichment alone, since nothing was swapped. on_unresolved What to do with a patch the inventory does not describe — one carrying no `acquisition_key`, one carrying a key the - inventory does not resolve to exactly one entry, or one - reaching outside every matching epoch. "raise" (the default) + inventory does not resolve to exactly one entry, one reaching + outside every matching epoch, or one with no instants to + resolve at because its time axis is not physical. A patch is + judged over its whole span, so one described at its start but + not at its end is undescribed. "raise" (the default) fails and names them, "warn" drops them and says so, and "drop" discards them silently, which is what an inventory deliberately covering part of an archive wants. @@ -1342,11 +1342,13 @@ def conform_to_inventory( Raises ------ PatchError - If a patch spans a change of *acquisition*. Its two halves - were recorded under different configurations, so no - subdivision makes it one honest patch, and `on_unresolved` - does not cover it: the inventory describes such a patch - twice rather than not at all. + If a patch spans a change of *acquisition*, or must be + subdivided but states no time step to find its samples with. + An acquisition change means the two halves were recorded + under different configurations, so no subdivision makes it + one honest patch, and `on_unresolved` does not cover it: the + inventory describes such a patch twice rather than not at + all. Examples -------- @@ -1368,22 +1370,11 @@ def conform_to_inventory( resolve_row_epochs, ) - if on_unresolved not in _VALID_ON_UNRESOLVED_CONFORM: - msg = ( - f"on_unresolved must be one of {_VALID_ON_UNRESOLVED_CONFORM}, " - f"got {on_unresolved!r}." - ) - raise ParameterError(msg) - if inventory is None and self._inventory is None: - msg = ( - "Spool.conform_to_inventory needs an inventory: pass one, or " - "attach one first with Spool.attach_inventory." - ) - raise ParameterError(msg) - new = ( - self.__class__(self) - if inventory is None - else self.attach_inventory(inventory) + new = self._with_inventory( + inventory, + on_unresolved, + _VALID_ON_UNRESOLVED_CONFORM, + "conform_to_inventory", ) source_rows, working = new._plan_frames() # The two frames are one relation split by column, so a row of @@ -1416,13 +1407,42 @@ def conform_to_inventory( ) return new._new_from_catalog(catalog) - def _restrict_to_rows(self, patch_ids) -> Self: - """Return the view holding only the named rows, in the same order.""" + def _with_inventory(self, inventory, on_unresolved, valid, method) -> Self: + """ + Return the spool an inventory verb works on, arguments checked. + + Both verbs take an inventory the same way — the attached one by + default, and one passed explicitly is attached as well — and both + police their own policy vocabulary before doing any work. Sharing + the entry keeps the two from drifting into saying it differently. + """ + if on_unresolved not in valid: + msg = f"on_unresolved must be one of {valid}, got {on_unresolved!r}." + raise ParameterError(msg) + if inventory is None and self._inventory is None: + msg = ( + f"Spool.{method} needs an inventory: pass one, or attach one " + "first with Spool.attach_inventory." + ) + raise ParameterError(msg) + if inventory is None: + return self.__class__(self) + return self.attach_inventory(inventory) + + def _restrict_to_rows(self, patch_ids, keep: bool = True) -> Self: + """ + Return the view holding the named rows, or all but them. + + Presentation order is the catalog's throughout, so this narrows + which rows a spool holds without saying anything about how they + come out. + """ ids = np.asarray(self._catalog._ordered_ids(), dtype=np.int64) - keep = np.isin(ids, np.asarray(patch_ids, dtype=np.int64)) - if keep.all(): + named = np.isin(ids, np.asarray(patch_ids, dtype=np.int64)) + mask = named if keep else ~named + if mask.all(): return self - return self._new_from_catalog(self._catalog.restrict(keep, ids=ids)) + return self._new_from_catalog(self._catalog.restrict(mask, ids=ids)) def _enrichment(self): """Return how this spool enriches, or None if it does not.""" diff --git a/dascore/proc/inventory.py b/dascore/proc/inventory.py index d01ee5be..6e111ce9 100644 --- a/dascore/proc/inventory.py +++ b/dascore/proc/inventory.py @@ -242,27 +242,29 @@ class RowEpochs(NamedTuple): cuts The instants inside the row at which its optical path changes; empty for a row which stays within one epoch of everything. - described - Whether the inventory resolves the row, over its whole span. conflict The instant at which the row's *acquisition* changes, or None. Subdividing cannot rescue this: the pieces would describe one patch recorded under two configurations, which is a file that should not exist rather than one to reconcile. context - What the row resolves to where it begins, or None where it is - undescribed. A row with no cuts and no conflict resolves to this - over its whole span. + What the row resolves to where it begins, or None where the + inventory does not describe the row over its whole span. A row + with no cuts and no conflict resolves to this throughout. """ cuts: tuple - described: bool conflict: Any context: ResolvedContext | None + @property + def described(self) -> bool: + """Whether the inventory resolves this row, over its whole span.""" + return self.context is not None + # What a row whose key names no entry at all knows about its epochs. -_NO_EPOCHS = RowEpochs((), False, None, None) +_NO_EPOCHS = RowEpochs((), None, None) def resolve_row_epochs(inventory, keys, starts, ends) -> list[RowEpochs]: @@ -285,7 +287,9 @@ def resolve_row_epochs(inventory, keys, starts, ends) -> list[RowEpochs]: keys Each row's acquisition_key. starts, ends - Each row's first and last instant. + Each row's first and last instant — the instants themselves, so + a row whose last instant falls exactly on a bound reaches into + the epoch that bound opens. Returns ------- @@ -316,8 +320,10 @@ def resolve_row_epochs(inventory, keys, starts, ends) -> list[RowEpochs]: ends_at = np.searchsorted(bounds, last, side="right") # A row with no instant of its own says nothing about which epoch # applies, and resolving at NaT holds every epoch effective; that - # is the whole inventory answering rather than one entry. - undated = np.isnat(first) | np.isnat(last) + # is the whole inventory answering rather than one entry. A row + # ending before it starts spans no epoch at all, which is the + # same nothing to resolve against, reached by a different route. + undated = np.isnat(first) | np.isnat(last) | (last < first) contexts: dict[int, ResolvedContext | None] = {} for position, (lo, hi) in enumerate(zip(starts_at, ends_at)): if undated[position]: @@ -346,26 +352,37 @@ def _try_resolve(inventory, key, when) -> ResolvedContext | None: return None +def _same(first, second) -> bool: + """ + Return whether two resolutions say the same thing. + + Identity first, because one inventory hands out the same object for + the same epoch and that costs nothing to check; only where the + objects differ is it worth dumping them to compare by value. What + matters to a patch is what the entry *says*, so an entity + re-registered unchanged across a bound is not a change — the same + call `_resolve_context` makes for a single patch. + """ + return first is second or first == second + + def _epoch_changes(resolved: list, boundaries) -> RowEpochs: """ Reduce a row's consecutive contexts to what changes between them. - Compared by identity: one inventory hands out the same objects for - the same epoch, so identity says exactly "the answer changed" — and - says it without dumping two model trees to find out. A bound the - answer does not change across is not a boundary this row crosses. + Only the acquisition and the optical path are compared, exactly as + `_resolve_context` compares them: the network and fiber array a + patch hangs from say nothing about it that its acquisition does not, + and a bound the answers do not change across is not a boundary this + row crosses at all. """ cuts = [] for previous, current, boundary in zip(resolved, resolved[1:], boundaries): - if ( - previous.network is not current.network - or previous.fiber_array is not current.fiber_array - or previous.acquisition is not current.acquisition - ): - return RowEpochs(tuple(cuts), True, boundary, resolved[0]) - if previous.optical_path is not current.optical_path: + if not _same(previous.acquisition, current.acquisition): + return RowEpochs(tuple(cuts), boundary, resolved[0]) + if not _same(previous.optical_path, current.optical_path): cuts.append(boundary) - return RowEpochs(tuple(cuts), True, None, resolved[0]) + return RowEpochs(tuple(cuts), None, resolved[0]) def get_attr_values(inventory, contexts, name: str) -> list: diff --git a/dascore/utils/chunk_plan.py b/dascore/utils/chunk_plan.py index cb73ae7c..b862b5f2 100644 --- a/dascore/utils/chunk_plan.py +++ b/dascore/utils/chunk_plan.py @@ -16,6 +16,7 @@ from __future__ import annotations import inspect +import math import warnings from dataclasses import dataclass, field from pathlib import Path @@ -1046,8 +1047,20 @@ def _snapped_cuts(cuts, start, step) -> list: span = to_float(step) out = [] for cut in cuts: - index = int(np.ceil(to_float(cut - start) / span)) - # Cuts fall strictly inside the row (see the docstring's + # The ratio is inexact — `to_float` rounds both operands and then + # the quotient — so it only starts the search, and the grid it + # describes settles it. `CoordRange._get_index`'s fudge factor is + # not enough here: at a million samples the error outgrows any + # fixed tolerance, and an index one too high puts the boundary + # sample in the epoch *before* the boundary, which is the one + # place it must not go. Comparing the values themselves is exact, + # and each loop steps at most once for any plausible error. + index = math.ceil(to_float(cut - start) / span) + while start + (index - 1) * step >= cut: + index -= 1 + while start + index * step < cut: + index += 1 + # Cuts sit above the row's minimum (see `build_subdivision_plan`'s # contract), so each one really does open a piece. assert index >= 1 # Two cuts inside one sample interval name one split: the epoch @@ -1074,9 +1087,10 @@ def build_subdivision_plan(df: pd.DataFrame, cuts, name: str) -> ChunkPlan: The relation to subdivide; one row per patch. cuts One sequence of cut values per row, in the row's own units, each - strictly inside that row's envelope. A cut opens a new piece at - the first sample at or after it, so a row with `n` distinct cuts - becomes at most `n + 1` outputs. + above that row's minimum and no greater than its maximum — a cut + on the maximum yields a one-sample final piece. A cut opens a new + piece at the first sample at or after it, so a row with `n` + distinct cuts becomes at most `n + 1` outputs. name The dimension being subdivided. @@ -1089,6 +1103,9 @@ def build_subdivision_plan(df: pd.DataFrame, cuts, name: str) -> ChunkPlan: min_name, max_name = f"{name}_min", f"{name}_max" step_name = f"{name}_step" assert {min_name, max_name, step_name}.issubset(df.columns) + # One entry per row, even where it is empty: a short sequence would + # drop the rows past its end from the plan, and so from the spool. + assert len(cuts) == len(df) df = _ensure_patch_id(df).reset_index(drop=True) positions, lows, highs, modified = [], [], [], [] for position, row_cuts in enumerate(cuts): diff --git a/tests/test_proc/test_proc_inventory.py b/tests/test_proc/test_proc_inventory.py index 71298bf9..7d0c3acb 100644 --- a/tests/test_proc/test_proc_inventory.py +++ b/tests/test_proc/test_proc_inventory.py @@ -45,6 +45,7 @@ UnitError, UnresolvedPatchError, ) +from dascore.proc.inventory import resolve_row_epochs @pytest.fixture(scope="module") @@ -2027,8 +2028,10 @@ def test_unstated_time_step_raises_naming_the_file(self, patch, inventory): uneven = patch.select(time=(0, 10), samples=True).update_coords(time=times) assert pd.isnull(dc.spool(uneven).get_contents()["time_step"]).all() split = _split_epochs(inventory, times[3], second={"name": "moved"}) - with pytest.raises(PatchError, match="no time step"): - dc.spool(uneven).conform_to_inventory(split) + spool = dc.spool(uneven) + named = re.escape(spool.get_contents()["source_path"].iloc[0]) + with pytest.raises(PatchError, match=f"no time step.*{named}"): + spool.conform_to_inventory(split) def test_a_row_with_no_instants_is_undescribed(self, patch, path_epochs): """ @@ -2134,9 +2137,12 @@ def test_warn_drops_and_names_them(self, patch, inventory): """The middle policy removes the rows but says which they were.""" other = patch.update_attrs(acquisition_key="DAS.R2D1..OTHER", tag="other") spool = dc.spool([patch, other]).attach_inventory(inventory) - with pytest.warns(UserWarning, match="does not describe"): + with pytest.warns(UserWarning, match="does not describe") as record: out = spool.conform_to_inventory(on_unresolved="warn") assert len(out) == 1 + # the point of warning at all is saying which patch was dropped + dropped = spool.get_contents()["source_path"].iloc[1] + assert dropped in str(record[0].message) def test_raise_is_the_default(self, patch, inventory): """Conforming to an inventory which covers less says so by default.""" @@ -2145,7 +2151,7 @@ def test_raise_is_the_default(self, patch, inventory): with pytest.raises(UnresolvedPatchError, match="does not describe"): spool.conform_to_inventory() - def test_len_matches_the_contents(self, patch, path_epochs): + def test_len_matches_the_contents(self, patch, inventory, path_epochs): """ Conforming is metadata work, so the count it reports is final. @@ -2153,12 +2159,13 @@ def test_len_matches_the_contents(self, patch, path_epochs): with what iteration will actually yield. """ other = patch.update_attrs(acquisition_key="DAS.R2D1..OTHER") - spool = dc.spool([patch, other]).attach_inventory(path_epochs) - for out in ( - spool.conform_to_inventory(on_unresolved="drop"), - dc.spool(other).conform_to_inventory(path_epochs, on_unresolved="drop"), - ): - assert len(out) == len(out.get_contents()) == len(list(out)) + subdivided = dc.spool([patch, other]).attach_inventory(path_epochs) + # the filtered route keeps a row rather than emptying the spool, + # so an agreement of zeroes cannot stand in for the real thing + filtered = dc.spool([patch, other]).attach_inventory(inventory) + for out, expected in ((subdivided, 2), (filtered, 1)): + out = out.conform_to_inventory(on_unresolved="drop") + assert len(out) == len(out.get_contents()) == len(list(out)) == expected def test_needs_an_inventory(self, patch): """With none attached and none passed there is nothing to conform to.""" @@ -2274,25 +2281,32 @@ def test_a_boundary_outside_the_patch_cuts_nothing(self, patch, inventory): def test_a_boundary_which_changes_nothing_cuts_nothing(self, patch, inventory): """ - An epoch bound the answer does not change across is not a boundary. + An epoch bound the answers do not change across is not a boundary. - Every stated time on the key's branch is a candidate, but only - the ones which actually resolve to a different path are cuts; - splitting on the rest would grow the spool for nothing. + Every stated time on the key's branch is a candidate, so a fiber + array re-registered mid-patch — renamed, say — puts a bound + inside the row while leaving its acquisition and optical path + saying exactly what they said before. Splitting there would grow + the spool for nothing, and refusing the patch would be worse. """ coord = patch.get_coord("time") + when = coord.min() + coord.step * 500 array = inventory.networks[0].fiber_arrays[0] - # the network's own epoch opens before the patch and closes after network = inventory.networks[0] - bounded = inventory.replace( + renamed = inventory.replace( network, network.new( - start_time=coord.min() - coord.step, - end_time=coord.max() + coord.step, - fiber_arrays=(array,), + fiber_arrays=( + array.new(end_time=when), + array.new(start_time=when, description="renamed"), + ) ), ).check() - assert len(dc.spool(patch).conform_to_inventory(bounded)) == 1 + # the bound really does fall inside the row + assert coord.min() < when <= coord.max() + assert len(dc.spool(patch).conform_to_inventory(renamed)) == 1 + # and enrich agrees, which is the point of comparing by value + assert patch.enrich(renamed).attrs.gauge_length == 10.0 def test_selection_reads_across_a_bound_which_changes_nothing( self, patch, inventory @@ -2305,17 +2319,22 @@ def test_selection_reads_across_a_bound_which_changes_nothing( have made the two disagree about the same patch. """ coord = patch.get_coord("time") + when = coord.min() + coord.step * 500 network = inventory.networks[0] - bounded = inventory.replace( + array = network.fiber_arrays[0] + renamed = inventory.replace( network, network.new( - start_time=coord.min() - coord.step, - end_time=coord.max() + coord.step, + fiber_arrays=( + array.new(end_time=when), + array.new(start_time=when, description="renamed"), + ) ), ).check() - spool = dc.spool(patch).attach_inventory(bounded) + assert coord.min() < when <= coord.max() + spool = dc.spool(patch).attach_inventory(renamed) assert len(spool.select(gauge_length=10.0)) == 1 - assert patch.enrich(bounded).attrs.gauge_length == 10.0 + assert patch.enrich(renamed).attrs.gauge_length == 10.0 def test_an_acquisition_change_raises(self, patch, inventory, off_grid_boundary): """ @@ -2331,8 +2350,11 @@ def test_an_acquisition_change_raises(self, patch, inventory, off_grid_boundary) second={"gauge_length": 20.0}, ) spool = dc.spool(patch).attach_inventory(split) + named = re.escape(spool.get_contents()["source_path"].iloc[0]) for policy in ("raise", "warn", "drop"): - with pytest.raises(PatchError, match="span a change of acquisition"): + with pytest.raises( + PatchError, match=f"change of acquisition.*{named} at .*" + ): spool.conform_to_inventory(on_unresolved=policy) @@ -2396,3 +2418,107 @@ def test_conform_over_a_directory_spool(self, tmp_path_factory, patch, inventory assert second.shape[1] == patch.shape[1] - 500 data = np.concatenate([first.data, second.data], axis=1) assert np.array_equal(data, patch.data) + + +class TestConformPartialCoverage: + """Rows the inventory describes for part of their span.""" + + def test_a_row_described_only_at_its_start_is_undescribed( + self, patch, inventory, off_grid_boundary + ): + """ + A patch is judged over its whole span, not where it begins. + + The acquisition lapses partway through with nothing after it, so + the tail of the patch has no entry. Keeping the described head + would answer a question the caller did not ask — they asked + about the patch — so the row is undescribed and `on_unresolved` + governs it. + """ + array = inventory.networks[0].fiber_arrays[0] + lapsed = inventory.replace( + array, + array.new( + acquisitions=(array.acquisitions[0].new(end_time=off_grid_boundary),) + ), + ).check() + spool = dc.spool(patch).attach_inventory(lapsed) + assert len(spool.conform_to_inventory(on_unresolved="drop")) == 0 + with pytest.raises(UnresolvedPatchError, match="does not describe"): + spool.conform_to_inventory() + + def test_a_path_which_lapses_divides_the_patch( + self, patch, inventory, off_grid_boundary + ): + """ + An optical path ending mid-patch is a change of path like any other. + + An acquisition with no optical path is a described acquisition — + it simply projects nothing along the fiber — so the piece after + the lapse is described once, as the piece before it is. This is + why a lapsed *path* subdivides where a lapsed *acquisition* + leaves the row undescribed: the acquisition is what makes a + patch describable at all. + """ + array = inventory.networks[0].fiber_arrays[0] + lapsed = inventory.replace( + array, + array.new( + optical_paths=(array.optical_paths[0].new(end_time=off_grid_boundary),) + ), + ).check() + out = dc.spool(patch).conform_to_inventory(lapsed) + assert len(out) == 2 + first, second = out.enrich(coords=True) + assert "zone" in first.coords.coord_map + assert "zone" not in second.coords.coord_map + + def test_many_undescribed_patches_are_summarized(self, patch, inventory): + """Naming every file of a mismatched archive helps nobody.""" + spool = dc.spool( + [ + patch.update_attrs(acquisition_key="DAS.R2D1..OTHER", tag=f"t{index}") + for index in range(7) + ] + ).attach_inventory(inventory) + with pytest.raises(UnresolvedPatchError, match=r"\(and 2 more\)"): + spool.conform_to_inventory() + + def test_a_row_ending_before_it_starts_resolves_to_nothing(self, patch, inventory): + """ + An envelope running backwards spans no epoch, so it answers none. + + Envelopes are value-ordered by construction, so this is a + malformed row rather than a reachable state; resolution says so + rather than failing on the empty span it computes. + """ + coord = patch.get_coord("time") + (epochs,) = resolve_row_epochs( + inventory, [patch.attrs.acquisition_key], [coord.max()], [coord.min()] + ) + assert not epochs.described and not epochs.cuts + + +class TestConformAndEnrichment: + """How conforming interacts with enrichment already set up.""" + + def test_conforming_keeps_enrichment(self, patch, path_epochs): + """The spool's own inventory is not a new one, so nothing is swapped.""" + spool = dc.spool(patch).enrich(path_epochs, coords=False) + out = spool.conform_to_inventory() + assert out._enrich_kwargs is not None + assert out[0].attrs.gauge_length == 10.0 + + def test_passing_an_inventory_clears_enrichment(self, patch, path_epochs): + """ + Passing one attaches it, and attaching always clears enrichment. + + Swapping the source underneath a configured enrichment would + rewrite every patch's metadata, so the new one has to be asked + for — the rule `attach_inventory` sets, which conform inherits + by taking its inventory the same way `enrich` does. + """ + spool = dc.spool(patch).enrich(path_epochs, coords=False) + out = spool.conform_to_inventory(path_epochs) + assert out._enrich_kwargs is None + assert "gauge_length" not in dict(out[0].attrs) diff --git a/tests/test_utils/test_chunk.py b/tests/test_utils/test_chunk.py index 2fab50fd..32b384ec 100644 --- a/tests/test_utils/test_chunk.py +++ b/tests/test_utils/test_chunk.py @@ -11,7 +11,11 @@ import dascore as dc from dascore.exceptions import ChunkError, ParameterError, UnitError from dascore.utils.chunk import get_intervals -from dascore.utils.chunk_plan import _normalize_chunk_units, build_chunk_plan +from dascore.utils.chunk_plan import ( + _normalize_chunk_units, + build_chunk_plan, + build_subdivision_plan, +) from dascore.utils.time import to_timedelta64 STARTTIME = np.datetime64("2020-01-03") @@ -533,3 +537,85 @@ def test_mixed_dtypes_upcast(self, sized_df): df.loc[df.index[0], "_dtype"] = "float32" plan = build_chunk_plan(df, time=dc.get_quantity("10 kB")) assert plan.params["size"]["partitions"][0]["dtype"] == "float64" + + +def _one_row_df(start, step, samples): + """A one-row relation over an evenly sampled time dimension.""" + return pd.DataFrame( + { + "time_min": [start], + "time_max": [start + step * (samples - 1)], + "time_step": [step], + "_patch_id": [0], + } + ) + + +class TestBuildSubdivisionPlan: + """Splitting each row of a relation at its own cut values.""" + + def test_uncut_rows_pass_through(self, contiguous_df): + """A row with no cuts is one unmodified output of the same span.""" + df = contiguous_df.assign(_patch_id=np.arange(len(contiguous_df))) + plan = build_subdivision_plan(df, [()] * len(df), "time") + assert len(plan.outputs) == len(df) + assert not plan.members["_modified"].any() + assert (plan.outputs["time_min"].values == df["time_min"].values).all() + + def test_pieces_partition_the_samples(self): + """The pieces meet exactly one step apart, with nothing between.""" + start, step = STARTTIME, np.timedelta64(10, "ms") + df = _one_row_df(start, step, 100) + cut = start + step * 40 + plan = build_subdivision_plan(df, [(cut,)], "time") + assert len(plan.outputs) == 2 + first, second = plan.outputs.iloc[0], plan.outputs.iloc[1] + assert second["time_min"] == cut + assert first["time_max"] == cut - step + assert plan.members["_modified"].all() + + @pytest.mark.parametrize("index", [3, 4001, 28294, 170275]) + def test_an_on_grid_cut_lands_on_its_own_sample(self, index): + """ + A cut sitting exactly on a sample opens the piece at that sample. + + The index is found by a float division, which is inexact — at + these step and index combinations the naive quotient rounds just + above the integer — so an unchecked `ceil` puts the boundary + sample one piece too early. Only a grid comparison settles it. + """ + start, step = STARTTIME, np.timedelta64(17060962, "ns") + df = _one_row_df(start, step, index + 10) + cut = start + step * index + plan = build_subdivision_plan(df, [(cut,)], "time") + assert plan.outputs["time_min"].iloc[1] == cut + + def test_a_cut_a_hair_past_a_sample_opens_at_the_next(self): + """ + The other direction: a cut just above a sample still snaps up. + + Far enough into a long-stepped row, one nanosecond is below the + precision of the ratio, so the float says the cut is *on* the + sample it is really just past. That sample belongs to the piece + before the cut, and the next one opens the piece after it. + """ + start, step = STARTTIME, np.timedelta64(1, "D").astype("timedelta64[ns]") + df = _one_row_df(start, step, 1100) + cut = start + step * 1000 + np.timedelta64(1, "ns") + plan = build_subdivision_plan(df, [(cut,)], "time") + assert plan.outputs["time_min"].iloc[1] == start + step * 1001 + + def test_a_cut_on_the_maximum_yields_one_sample(self): + """A row ending exactly on a boundary keeps its last sample apart.""" + start, step = STARTTIME, np.timedelta64(10, "ms") + df = _one_row_df(start, step, 50) + cut = df["time_max"].iloc[0] + plan = build_subdivision_plan(df, [(cut,)], "time") + assert len(plan.outputs) == 2 + assert plan.outputs["time_min"].iloc[1] == plan.outputs["time_max"].iloc[1] + + def test_one_cut_sequence_per_row(self): + """A short sequence would silently drop rows from the plan.""" + df = _one_row_df(STARTTIME, np.timedelta64(10, "ms"), 10) + with pytest.raises(AssertionError): + build_subdivision_plan(pd.concat([df, df]), [()], "time") From 6ca77f58bbaa191efc8c6dc1a392ea35a7381d77 Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Wed, 12 Aug 2026 19:22:53 +0200 Subject: [PATCH 3/3] Settle the cut order, and zip in step Which cuts a row has is a set, not a sequence, but the pieces are read off consecutive pairs -- so an unordered one would describe envelopes running backwards rather than raise. Sort them where they are computed, so no caller can get it wrong. The epoch walk's zips now pair sequences of equal length explicitly. The three-way one needed `resolved[:-1]`, not just `strict=True`: there is one boundary between each consecutive pair, so the untrimmed sequence was always one longer. --- dascore/proc/inventory.py | 8 ++++++-- dascore/utils/chunk_plan.py | 12 ++++++++---- tests/test_proc/test_proc_inventory.py | 9 +++++++-- 3 files changed, 21 insertions(+), 8 deletions(-) diff --git a/dascore/proc/inventory.py b/dascore/proc/inventory.py index 6e111ce9..2cd8ccae 100644 --- a/dascore/proc/inventory.py +++ b/dascore/proc/inventory.py @@ -325,7 +325,7 @@ def resolve_row_epochs(inventory, keys, starts, ends) -> list[RowEpochs]: # same nothing to resolve against, reached by a different route. undated = np.isnat(first) | np.isnat(last) | (last < first) contexts: dict[int, ResolvedContext | None] = {} - for position, (lo, hi) in enumerate(zip(starts_at, ends_at)): + for position, (lo, hi) in enumerate(zip(starts_at, ends_at, strict=True)): if undated[position]: continue for epoch in range(int(lo), int(hi) + 1): @@ -377,7 +377,11 @@ def _epoch_changes(resolved: list, boundaries) -> RowEpochs: row crosses at all. """ cuts = [] - for previous, current, boundary in zip(resolved, resolved[1:], boundaries): + # One boundary between each consecutive pair, so the three walk in + # step -- `resolved[1:]` alone would leave the first sequence longer. + for previous, current, boundary in zip( + resolved[:-1], resolved[1:], boundaries, strict=True + ): if not _same(previous.acquisition, current.acquisition): return RowEpochs(tuple(cuts), boundary, resolved[0]) if not _same(previous.optical_path, current.optical_path): diff --git a/dascore/utils/chunk_plan.py b/dascore/utils/chunk_plan.py index b862b5f2..92936a74 100644 --- a/dascore/utils/chunk_plan.py +++ b/dascore/utils/chunk_plan.py @@ -1067,7 +1067,10 @@ def _snapped_cuts(cuts, start, step) -> list: # between them covers no sample of this row. if (value := start + index * step) not in out: out.append(value) - return out + # Which cuts a row has is a set, not a sequence — but the pieces are + # read off consecutive pairs, so an unordered one would describe + # envelopes running backwards rather than an error. + return sorted(out) def build_subdivision_plan(df: pd.DataFrame, cuts, name: str) -> ChunkPlan: @@ -1088,9 +1091,10 @@ def build_subdivision_plan(df: pd.DataFrame, cuts, name: str) -> ChunkPlan: cuts One sequence of cut values per row, in the row's own units, each above that row's minimum and no greater than its maximum — a cut - on the maximum yields a one-sample final piece. A cut opens a new - piece at the first sample at or after it, so a row with `n` - distinct cuts becomes at most `n + 1` outputs. + on the maximum yields a one-sample final piece. Order does not + matter. A cut opens a new piece at the first sample at or after + it, so a row with `n` distinct cuts becomes at most `n + 1` + outputs. name The dimension being subdivided. diff --git a/tests/test_proc/test_proc_inventory.py b/tests/test_proc/test_proc_inventory.py index 7d0c3acb..3562ab21 100644 --- a/tests/test_proc/test_proc_inventory.py +++ b/tests/test_proc/test_proc_inventory.py @@ -1940,7 +1940,7 @@ def _split_epochs(inventory, when, *, acquisitions=False, second=None): def _pieces(spool): """The (min, max) time envelope of each patch a spool presents.""" contents = spool.get_contents() - return list(zip(contents["time_min"], contents["time_max"])) + return list(zip(contents["time_min"], contents["time_max"], strict=True)) @pytest.fixture(scope="module") @@ -1957,7 +1957,12 @@ def path_epochs(inventory, off_grid_boundary): class TestConformBoundaryPolicy: - """Where a subdivided patch is cut, and which piece each sample joins.""" + """ + Where a subdivided patch is cut, and which piece each sample joins. + + Including the rows which offer no usable boundary at all — no + instants to place one against, or no identity to look one up with. + """ def test_split_is_lossless(self, patch, path_epochs): """