diff --git a/dascore/core/spool.py b/dascore/core/spool.py index 6c30d9cf..827e692d 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,122 @@ def _copy_public_dataframe(frame: pd.DataFrame) -> pd.DataFrame: _VALID_ON_UNRESOLVED = ("warn", "raise", "ignore") +# 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 = ( "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 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. + """ + 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"]) + 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)}. {advice}" + ) + 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 — 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}" + 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: 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]}" + 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. @@ -876,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.""" @@ -994,12 +1101,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 +1108,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 +1171,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 +1235,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 ---------- @@ -1188,26 +1290,159 @@ 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}." - ) + new = self._with_inventory( + inventory, on_unresolved, _VALID_ON_UNRESOLVED, "enrich" + ) + new._enrich_kwargs = enrich_kwargs + 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 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 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 — 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, 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. + + Raises + ------ + PatchError + 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 + -------- + >>> 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, + ) + + 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 + # 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 _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 = ( - "Spool.enrich needs an inventory: pass one, or attach one " + f"Spool.{method} 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._enrich_kwargs = enrich_kwargs - new._on_unresolved = on_unresolved - return new + 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) + 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(mask, ids=ids)) def _enrichment(self): """Return how this spool enriches, or None if it does not.""" @@ -1224,7 +1459,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..2cd8ccae 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,76 @@ 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. + 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 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 + 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((), 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 — the instants themselves, so + a row whose last instant falls exactly on a bound reaches into + the epoch that bound opens. + + Returns + ------- + A list of `RowEpochs`, one per row. + """ frame = pd.DataFrame( { "key": np.asarray(keys, dtype=object), @@ -230,7 +302,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 +312,83 @@ 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) + # 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 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, strict=True)): + 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 _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. + + 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 = [] + # 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): + cuts.append(boundary) + return RowEpochs(tuple(cuts), 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..92936a74 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 @@ -1030,3 +1031,116 @@ 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: + # 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 + # between them covers no sample of this row. + if (value := start + index * step) not in out: + out.append(value) + # 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: + """ + 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 + above that row's minimum and no greater than its maximum — a cut + 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. + + 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) + # 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): + 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..3562ab21 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, @@ -43,6 +45,7 @@ UnitError, UnresolvedPatchError, ) +from dascore.proc.inventory import resolve_row_epochs @pytest.fixture(scope="module") @@ -1916,3 +1919,611 @@ 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"], strict=True)) + + +@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. + + 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): + """ + 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"}) + 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): + """ + 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") 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.""" + 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, inventory, 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") + 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.""" + 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 answers do not change across is not a boundary. + + 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] + network = inventory.networks[0] + renamed = inventory.replace( + network, + network.new( + fiber_arrays=( + array.new(end_time=when), + array.new(start_time=when, description="renamed"), + ) + ), + ).check() + # 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 + ): + """ + 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") + when = coord.min() + coord.step * 500 + network = inventory.networks[0] + array = network.fiber_arrays[0] + renamed = inventory.replace( + network, + network.new( + fiber_arrays=( + array.new(end_time=when), + array.new(start_time=when, description="renamed"), + ) + ), + ).check() + 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(renamed).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) + named = re.escape(spool.get_contents()["source_path"].iloc[0]) + for policy in ("raise", "warn", "drop"): + with pytest.raises( + PatchError, match=f"change of acquisition.*{named} at .*" + ): + 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) + + +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")