Keep the Febus G1 sample window and fix two coordinate bugs it exposed

dascore/io/febus/core.py

@@ -188,7 +188,12 @@ def read(self, resource: TextReader, **kwargs) -> dc.BaseSpool:
 
 
 class FebusMTXH5V1(FiberIO):
-    """HDF5 format used by Febus for storing Brillouin spectra."""
+    """
+    HDF5 format used by Febus for storing Brillouin spectra.
+
+    As with the BSL files, ``time`` holds the start of each acquisition
+    window and the non-dimensional ``sample_span`` coord holds its length.
+    """
 
     name = "febus_mtx_h5"
     preferred_extensions = ("h5", "hdf5")
@@ -246,7 +251,13 @@ def read(
 
 
 class FebusBSLH5V1(FiberIO):
-    """HDF5 format used by Febus G1 for storing BSL strain files."""
+    """
+    HDF5 format used by Febus G1 for storing BSL strain files.
+
+    Samples are not instantaneous; each one covers an acquisition window.
+    The ``time`` coord holds the start of that window and the non-dimensional
+    ``sample_span`` coord, mapped to ``time``, holds how long it ran.
+    """
 
     name = "febus_bsl_h5"
     preferred_extensions = ("h5", "hdf5")

dascore/io/febus/g1utils.py

@@ -22,6 +22,10 @@
 _MTX_DIMS = ("time", "distance", "frequency")
 _BSL_DIMS = ("time", "distance")
 _BSL_H5_MODE_STRAIN = 1
+# The root "start_time"/"end_time" attrs are bit-identical to the first
+# start_times and last end_times entries, so they only restate the time
+# datasets as untyped epoch floats. "formatVersion" is already reported as
+# source_version. All three are deliberately absent here.
 _G1_H5_ATTR_MAP = {
     "acq_res": "acq_res",
     "ampliPower": "ampli_power",
@@ -31,7 +35,6 @@
     "fiberFrom": "fiber_from",
     "fiberLength": "fiber_length",
     "fiberTo": "fiber_to",
-    "formatVersion": "format_version",
     "freq_fiber": "freq_fiber",
     "freq_offset": "freq_offset",
     "freq_offset_abs": "freq_offset_abs",
@@ -41,8 +44,6 @@
     "sampling_resolution": "sampling_resolution",
     "signal_size": "signal_size",
     "spatial_resolution": "spatial_resolution",
-    "start_time": "start_time",
-    "end_time": "end_time",
     "zoneCount": "zone_count",
     "zones": "zones",
     "febusDataKind": "febus_data_kind",
@@ -190,13 +191,30 @@ def _coord(values, units=None):
         return get_exact_coord(values, units=units)
 
     extra_coords = {} if extra_coords is None else extra_coords
-    time = _coord(dc.to_datetime64(resource["start_times"][...]))
+    starts = resource["start_times"][...]
+    ends = resource["end_times"][...]
+    if ends.shape != starts.shape:
+        msg = (
+            f"start_times has shape {starts.shape} but end_times has "
+            f"{ends.shape}; the file is truncated or still being written."
+        )
+        raise ValueError(msg)
+    time = _coord(dc.to_datetime64(starts))
+    # Each sample covers a window rather than being instantaneous, so keep how
+    # long it ran. The span is differenced off the raw arrays rather than
+    # stored as end_times: starts and ends are each near-regular and snap to
+    # slightly different steps, so subtracting the two built coords would turn
+    # the jitter into a linear drift. Built exactly, and ignoring `snap`,
+    # because that jitter is the signal -- though a span array regular enough
+    # to look like a range is still normalized to one by the coord manager.
+    sample_span = get_exact_coord(dc.to_timedelta64(ends - starts))
     distance = _coord(resource["distances"][...], units="m")
     temperature = _coord(resource["temperatures"][...], units="°C")
     coords = {
         "time": time,
         "distance": distance,
         "temperature": ("time", temperature),
+        "sample_span": ("time", sample_span),
         **extra_coords,
     }
     return dc.get_coord_manager(coords, dims=dims)

dascore/io/index/backend.py

@@ -994,7 +994,11 @@ def _time_objects(ns_series: pd.Series, flavor: str) -> np.ndarray:
             for mask, flavor in time_flavors:
                 if mask.any():
                     values[mask] = _time_objects(coords[ns_col][mask], flavor)
-            coords[out_col] = values
+            # Assigning the bare array lets pandas re-infer a dtype; a result
+            # whose numeric coords all have null envelopes holds only
+            # Timestamp/Timedelta and None, which would infer a temporal dtype
+            # and silently turn those numeric nulls into NaT.
+            coords[out_col] = pd.Series(values, index=coords.index, dtype=object)
         # Summary-only definitions are useful for indexing/dedup but cannot
         # prove coordinate value identity for merge grouping.
         coords["_key"] = coords["def_key"].where(coords["fingerprint"].notna(), None)

dascore/utils/coordmanager.py

@@ -117,7 +117,7 @@ def _get_merged_coords(managers, coords_to_merge):
             if dim_coord is not None and coord_name == dim:
                 out[coord_name] = (managers[0].dim_map[dim], dim_coord)
                 continue
-            merge_coords = [x.coord_map[dim] for x in managers]
+            merge_coords = [x.coord_map[coord_name] for x in managers]
             axis = managers[0].dim_map[coord_name].index(dim)
             if len(units := {x.units for x in merge_coords}) != 1:
                 # TODO: we might try to convert all the units to a common
@@ -127,9 +127,12 @@ def _get_merged_coords(managers, coords_to_merge):
                     f"share the same units. Units found are: {set(units)}"
                 )
                 raise CoordMergeError(msg)
-            snap_coords = _snap_coords(merge_coords)
-            data = [x.data for x in snap_coords]
-            dims = managers[0].dim_map[dim]
+            # Only the dimension coordinate defines contiguity, so only it is
+            # snapped; coords merely associated with dim just follow along.
+            if coord_name == dim:
+                merge_coords = _snap_coords(merge_coords)
+            data = [x.data for x in merge_coords]
+            dims = managers[0].dim_map[coord_name]
             new_data = np.concatenate(data, axis=axis)
             # raw value concatenation loses the coord's units; reattach
             # the (verified common) units so the merge stays unit-true

tests/test_io/test_common_io.py

@@ -152,7 +152,6 @@
         "demod_data_type",
         "dtype",
         "end_frequency",
-        "end_time",
         "epsg_code",
         "event_depth_km",
         "event_id",
@@ -216,7 +215,6 @@
         "source",
         "spatial_resolution",
         "start_frequency",
-        "start_time",
         "temperature",
         "time_decimation_filter",
         "trace_count",

tests/test_io/test_febus/test_febusbsl.py

@@ -12,7 +12,7 @@
 import dascore as dc
 from dascore.constants import STORAGE_PROVENANCE_ATTRS
 from dascore.io.febus import FebusBSLH5V1
-from dascore.io.febus.g1utils import _get_bsl_attrs
+from dascore.io.febus.g1utils import _get_bsl_attrs, _get_g1_h5_base_coords
 from dascore.utils.downloader import fetch
 
 BSL_NAME = "febusg1_C2_2026-06-03T17.18.13+0200.bsl.h5"
@@ -100,6 +100,78 @@ def test_time_coord(self, bsl_patch):
         assert time.step is None
         assert bsl_patch.summary.get_coord_summary("time").step is None
 
+    def test_sample_span_coord(self, bsl_path, bsl_patch):
+        """Each sample covers a window; its exact length should be kept."""
+        with h5py.File(bsl_path) as fi:
+            expected = fi["end_times"][...] - fi["start_times"][...]
+        span = bsl_patch.get_coord("sample_span")
+        assert bsl_patch.coords.dim_map["sample_span"] == ("time",)
+        assert np.array_equal(span.values, dc.to_timedelta64(expected))
+
+    def test_sample_span_differenced_before_snapping(self):
+        """Spans come off the raw arrays, not from subtracting built coords.
+
+        Starts and ends are each near-regular and snap to slightly different
+        steps, so differencing the two coords would show a linear drift where
+        the file has jitter.
+        """
+        starts = np.array([0.0, 1.0, 2.0, 3.0])
+        ends = starts + np.array([0.9, 1.2, 0.8, 1.1])
+        coords = _get_g1_h5_base_coords(
+            {
+                "start_times": starts,
+                "end_times": ends,
+                "distances": np.arange(2, dtype=np.float64),
+                "temperatures": np.zeros(4, dtype=np.float64),
+            },
+            dims=("time", "distance"),
+        )
+        span = coords.coord_map["sample_span"]
+        assert np.array_equal(span.values, dc.to_timedelta64(ends - starts))
+        # the snapped time coord would have given a constant span
+        assert len(np.unique(span.values)) > 1
+
+    def test_mismatched_time_dataset_lengths_raise(self, bsl_path, tmp_path):
+        """A half-written file should fail loudly, not broadcast to garbage."""
+        new_path = tmp_path / bsl_path.name
+        shutil.copy2(bsl_path, new_path)
+        with h5py.File(new_path, "a") as fi:
+            ends = fi["end_times"][...]
+            del fi["end_times"]
+            fi.create_dataset("end_times", data=ends[:1])
+        with pytest.raises(ValueError, match="truncated or still being written"):
+            self.parser.read(new_path)
+
+    def test_redundant_time_attrs_absent(self, bsl_patch):
+        """Neither the epoch-float time attrs nor format_version reach attrs."""
+        names = set(dict(bsl_patch.attrs))
+        assert not names & {"start_time", "end_time", "format_version"}
+
+    def test_dropped_time_attrs_lose_nothing(self, bsl_path, bsl_patch):
+        """What the dropped attrs said is still recoverable from the coords.
+
+        This is what justifies removing them rather than retyping them.
+        """
+        with h5py.File(bsl_path) as fi:
+            start_attr = float(fi.attrs["start_time"][0])
+            end_attr = float(fi.attrs["end_time"][0])
+        time = bsl_patch.get_coord("time")
+        span = bsl_patch.get_coord("sample_span")
+        assert time.min() == dc.to_datetime64(start_attr)
+        # The end rebuilds through two independent float-seconds-to-ns
+        # roundings, so it lands within a sample of the stored attr rather
+        # than on it; the file's own epoch floats are only good to ~238 ns
+        # here anyway.
+        rebuilt = time.max() + span.values[-1]
+        assert abs(rebuilt - dc.to_datetime64(end_attr)) < np.timedelta64(1, "us")
+
+    def test_select_slices_sample_span(self, bsl_path, bsl_patch):
+        """Selecting on time should carry the associated span along."""
+        time = bsl_patch.get_coord("time")
+        out = self.parser.read(bsl_path, time=(time.values[3], time.values[8]))[0]
+        expected = bsl_patch.get_coord("sample_span").values[3:9]
+        assert np.array_equal(out.get_coord("sample_span").values, expected)
+
     def test_select(self, bsl_path, bsl_patch):
         """Partial reads should reduce coords and data consistently."""
         time = bsl_patch.get_coord("time")

tests/test_io/test_febus/test_febusg1.py

@@ -157,10 +157,16 @@ def test_read(self, mtx_h5_path):
         assert patch.shape == (68, 100, 128)
         assert patch.attrs.data_category == "DSS"
         assert patch.attrs.data_type == "brillouin_spectrum"
-        assert patch.attrs.format_version == 1
         assert patch.attrs.fiber_from == 50
         assert "temperature" in patch.coords.coord_map
         assert patch.coords.dim_map["temperature"] == ("time",)
+        assert patch.coords.dim_map["sample_span"] == ("time",)
+
+    def test_format_version_not_duplicated_in_attrs(self, mtx_h5_path):
+        """The file version is reported as the format version, not as an attr."""
+        patch = dc.read(mtx_h5_path)[0]
+        assert "format_version" not in dict(patch.attrs)
+        assert dc.scan(mtx_h5_path)[0].source_version == FebusMTXH5V1.version
 
     def test_read_preserves_mtx_array_order(self, tmp_path):
         """The patch data should match the MTX array stored in the file."""

tests/test_io/test_index/test_index_edge_cases.py

@@ -141,6 +141,62 @@ def _summary(gain, path):
         finally:
             back.close()
 
+    def test_stepless_numeric_coord_keeps_null_step(self, tmp_path):
+        """A numeric coord with no step stays NaN rather than an int sentinel.
+
+        When no numeric coord in the result carries a step, the envelope
+        column holds only timedeltas and nulls; inferring a dtype there
+        turned those nulls into NaT, and then into int64 min.
+        """
+        # the time coord's step is what puts a Timedelta in the shared envelope
+        # column; without it there is nothing for pandas to infer a temporal
+        # dtype from, and the bug cannot occur. Stated here rather than
+        # inherited from a helper default so the trigger cannot drift away.
+        time_coord = _time_coord("2024-01-01T00:00:00", 60, step_s=0.004)
+        assert time_coord["step"] is not None
+        summary = PatchSummary(
+            attrs={"tag": "t"},
+            coords={
+                "time": time_coord,
+                # both numeric coords are stepless, so nothing anchors the
+                # envelope column to a numeric dtype
+                "distance": {
+                    "dtype": "float64",
+                    "min": 0.0,
+                    "max": 10.0,
+                    "step": None,
+                    "units": "m",
+                    "dims": ("distance",),
+                    "len": 11,
+                },
+                "temperature": {
+                    "dtype": "float64",
+                    "min": 20.0,
+                    "max": 25.0,
+                    "step": None,
+                    "units": "degC",
+                    "dims": ("time",),
+                    "len": 15000,
+                },
+            },
+            dims=("time", "distance"),
+            shape=(15000, 11),
+            dtype="float32",
+            source_path="a.h5",
+            source_format="X",
+            source_version="1",
+        )
+        back = get_backend(tmp_path / "i.sqlite3")
+        try:
+            back.write_sources(s2r([summary]))
+            df = back.query()
+            step = df["temperature_step"]
+            # int64 min is what a NaT in this column decays to
+            assert step.dtype == np.dtype("float64")
+            assert step.isna().all()
+        finally:
+            back.close()
+
     def test_reopen_missing_meta_row(self, tmp_path):
         """An index whose meta row was lost is rejected on reopen."""
         path = tmp_path / "i.sqlite3"

tests/test_utils/test_coordmanager_utils.py

@@ -51,6 +51,27 @@ def test_merge_simple(self, cm_basic):
         assert new_time.min() == time.min()
         assert new_time.max() == cm2.coord_map["time"].max()
 
+    def test_merge_keeps_associated_coord_values(self, cm_basic):
+        """A coord along the merge dim keeps its own values, not the dim's."""
+        size = cm_basic.shape[cm_basic.get_axis("time")]
+        cm1 = cm_basic.update_coords(quality=("time", np.arange(size)))
+        time = cm1.coord_map["time"]
+        # each manager gets its own quality values; identical ones could not
+        # tell a real merge from the first manager's values used twice
+        cm2 = self._get_offset_coord_manager(cm1, time=time.step * 1.1)
+        cm2 = cm2.update_coords(quality=("time", np.arange(size) + 100))
+        # a tolerance the offset above lands inside, so snapping runs: the
+        # dim coord is snapped and quality, having no step, must be left alone
+        out = merge_coord_managers([cm1, cm2], dim="time", snap_tolerance=1.3)
+        quality = out.coord_map["quality"]
+        expected = np.concatenate([np.arange(size), np.arange(size) + 100])
+        # the dim coord's values used to be substituted here, which for a
+        # datetime time dim also silently changed the coord's dtype
+        assert quality.dtype == cm1.coord_map["quality"].dtype
+        assert np.array_equal(quality.values, expected)
+        assert out.dim_map["quality"] == ("time",)
+        assert out.coord_map["time"].shape == quality.shape
+
     def test_merge_offset_close_no_snap(self, cm_basic):
         """When the coordinate don't line up, it should produce monotonic Coord."""
         cm1 = cm_basic