Skip to content

Extract interval and table machinery from the inventory modules #911

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
d-chambers merged 2 commits into from
Aug 15, 2026
Merged

Extract interval and table machinery from the inventory modules #911

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4bd6cc1
Extract interval and table machinery from the inventory modules
d-chambers Aug 15, 2026
da46eda
Merge dev into annotations-1
d-chambers Aug 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 3 additions & 4 deletions dascore/core/_spool_inventory.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,8 +33,6 @@
VALID_COORDINATE_LABELS,
Inventory,
ResolvedContext,
_annotation_kind,
interval_masks,
)
from dascore.core.inventory_loader import BLESSED_NAME, find_inventory
from dascore.exceptions import (
Expand All @@ -47,6 +45,7 @@
UnresolvedPatchError,
)
from dascore.units import get_quantity_str
from dascore.utils.intervals import interval_masks, value_kind
from dascore.utils.misc import iterate

# One vocabulary for both fiber verbs. What the quiet option leaves
Expand Down Expand Up @@ -625,7 +624,7 @@ def _get_annotation_coord(path, group, distances):
items = [x for x in path.annotations if x.group == group]
if not items:
return None
kind = _annotation_kind(items[0].value)
kind = value_kind(items[0].value)
intervals = [x.interval for x in items]
values = [x.value for x in items]
return _fill_from_intervals(distances, intervals, values, kind)
Expand All @@ -648,7 +647,7 @@ def _get_track_coord(path, track, field, distances):
# inventory defines nothing here, and on_missing then rules --
# rather than handing back a coordinate that is blank throughout.
return None
kinds = {_annotation_kind(x) for x in values if not is_unset(x)}
kinds = {value_kind(x) for x in values if not is_unset(x)}
kind = kinds.pop() if len(kinds) == 1 else "string"
filled = _fill_from_intervals(distances, intervals, values, kind)
if units := _TRACK_FIELD_UNITS.get(f"{track}.{field}"):
Expand Down
100 changes: 14 additions & 86 deletions dascore/core/inventory.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,13 @@
TimeRangedModel,
UnitQuantity,
)
from dascore.utils.intervals import (
clip_intervals,
interval_masks,
intervals_overlap,
normalize_value,
value_kind,
)
from dascore.utils.mapping import FrozenDict
from dascore.utils.misc import (
check_code,
Expand Down Expand Up @@ -107,18 +114,8 @@


def _annotation_value(value):
"""
Normalize an annotation value so its Python type survives validation.

Numpy scalars are unwrapped: pydantic's smart union resolves every numpy
scalar to float, which would turn a mask element into a numeric group.
"""
if isinstance(value, np.generic):
value = value.item()
if isinstance(value, float) and not np.isfinite(value):
msg = f"Annotation value must be finite; got {value}."
raise InvalidInventoryError(msg)
return value
"""Normalize an annotation value so its Python type survives validation."""
return normalize_value(value, error=InvalidInventoryError)


# The value kind decides an annotation group's shape, so it must be exact.
Expand Down Expand Up @@ -574,31 +571,6 @@ def interpolate(self, distances) -> np.ndarray:
return out


def interval_masks(values, intervals) -> list[np.ndarray]:
"""
Return, per interval, the mask of values that interval covers.

Coverage is half-open, ``[start, end)``, with one exception: the end of
a coverage run belongs to the interval ending there when no half-open
interval claims it, so the last point of a run is not left out. Point
markers (equal start and end) cover nothing.
"""
values = np.asarray(values, dtype=float)
spans = [(lo, hi) for lo, hi in intervals]
claimed = np.zeros(len(values), dtype=bool)
for lo, hi in spans:
if lo < hi:
claimed |= (values >= lo) & (values < hi)
out = []
for lo, hi in spans:
if lo >= hi: # a point marker covers nothing
out.append(np.zeros(len(values), dtype=bool))
continue
mask = (values >= lo) & (values < hi)
out.append(mask | ((values == hi) & ~claimed))
return out


class _IntervalModel(InventoryModel):
"""
Base for items covering the half-open interval [start, end) of optical
Expand Down Expand Up @@ -1062,27 +1034,6 @@ def _times_equal(time1, time2) -> bool:
return bool(time1 == time2)


def _annotation_kind(value) -> str:
"""Return the value kind which decides an annotation group's shape."""
if isinstance(value, bool): # bool before int; bool is an int subclass
return "boolean"
if isinstance(value, str):
return "string"
return "numeric"


def _intervals_overlap(intervals: list[tuple[float, float]]) -> tuple | None:
"""Return the first overlapping pair of half-open intervals, or None.

Empty (point) intervals cover nothing and cannot overlap.
"""
ordered = sorted(x for x in intervals if x[0] < x[1])
for first, second in itertools.pairwise(ordered):
if second[0] < first[1]:
return first, second
return None


class OpticalPath(TimeRangedModel):
"""
Continuous optical path described by independent tracks.
Expand Down Expand Up @@ -1203,7 +1154,7 @@ def check(self, tolerance: float = 1e-9) -> Self:
if len(dims) > 1:
errors.append(_MIXED_DIMS_MSG.format(dims=sorted(dims)))
for name, spans in (("geometry", geo_spans), ("coupling", coup_spans)):
overlap = _intervals_overlap(spans)
overlap = intervals_overlap(spans)
if overlap is not None:
errors.append(
f"Overlapping {name} intervals {overlap[0]} and "
Expand All @@ -1228,7 +1179,7 @@ def _check_annotation_groups(self) -> list[str]:
"coordinate, a typed track, or a coordinate label."
)
for group, items in groups.items():
kinds = {_annotation_kind(x.value) for x in items}
kinds = {value_kind(x.value) for x in items}
if len(kinds) > 1:
errors.append(
f"Annotation group {group!r} mixes {sorted(kinds)} values; "
Expand All @@ -1237,7 +1188,7 @@ def _check_annotation_groups(self) -> list[str]:
continue
if kinds == {"boolean"}: # membership groups may overlap
continue
overlap = _intervals_overlap([x.interval for x in items])
overlap = intervals_overlap([x.interval for x in items])
if overlap is not None:
errors.append(
f"Overlapping intervals {overlap[0]} and {overlap[1]} in "
Expand Down Expand Up @@ -1295,8 +1246,8 @@ def select(self, *, distance: tuple[float | None, float | None]) -> Self:
)
)
outer = self.end_distance
coupling = _clip_intervals(self.coupling, lo, hi, outer)
annotations = _clip_intervals(self.annotations, lo, hi, outer)
coupling = clip_intervals(self.coupling, lo, hi, outer)
annotations = clip_intervals(self.annotations, lo, hi, outer)
return self.model_copy(
update={
"start_distance": lo,
Expand Down Expand Up @@ -1422,29 +1373,6 @@ def shift_item(item):
)


def _clip_intervals(items, lo: float, hi: float, outer: float | None = None) -> list:
"""
Clip interval items to [lo, hi), dropping those left with no coverage.

Point markers cover nothing but are not nothing: they survive when they
fall inside the clip, or on its outermost included endpoint.
"""
out = []
for item in items:
start, end = item.interval
if start == end:
if lo <= start < hi or (outer is not None and start == hi == outer):
out.append(item)
continue
new_lo, new_hi = max(start, lo), min(end, hi)
if new_hi <= new_lo:
continue
out.append(
item.model_copy(update={"start_distance": new_lo, "end_distance": new_hi})
)
return out


class Response(InventoryModel):
"""Station-specific response model associated with a channel."""

Expand Down
Loading
Loading