fix(waterdata.xarray): handle NaT times and mixed units in the converter

Two correctness fixes in _build_timeseries from review:

- NaT time: a row whose time fails to parse (errors="coerce" -> NaT) could
  not index the array and was silently swallowed by to_xarray (and an
  all-NaT group crashed). Drop such rows explicitly with a warning, and
  return an empty Dataset if nothing parseable remains -- no more silent
  data loss.

- Mixed units: unit_of_measure was collapsed with _first over a
  (parameter_code, statistic_id) group that can span multiple units (e.g.
  the same parameter reported in different units across sites). A variable
  can carry only one units attr, so warn when a group spans multiple units
  instead of silently labeling every value with the first.

Adds tests for the bad-time drop, all-bad-time empty result, and the
mixed-unit warning.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: thodson-usgs

dataretrieval/waterdata/xarray.py

@@ -242,11 +242,26 @@ def _build_timeseries(
     ).dt.tz_localize(None)
     work["value"] = _pd.to_numeric(work["value"], errors="coerce")
 
+    # A NaT time can't index the array and would be silently swallowed by
+    # ``to_xarray`` (or crash an all-NaT group). Drop such rows explicitly and
+    # say so, rather than losing observations without a trace.
+    n_before = len(work)
+    work = work[work["time"].notna()]
+    dropped = n_before - len(work)
+    if dropped:
+        _warnings.warn(
+            f"dropped {dropped} row(s) with an unparseable or missing time.",
+            stacklevel=3,
+        )
+    if work.empty:
+        return _empty_dataset(service, base_meta)
+
     datasets, used = [], set()
     for _, group in work.groupby(group_cols, dropna=False):
         pcode = _first(group["parameter_code"]) if "parameter_code" in group else None
         stat = _first(group["statistic_id"]) if "statistic_id" in group else None
-        unit = _first(group["unit_of_measure"]) if has_unit else None
+        group_units = group["unit_of_measure"].dropna().unique() if has_unit else ()
+        unit = group_units[0] if len(group_units) else None
         desc = series_meta.get(str(pcode), {}) if pcode is not None else {}
 
         name = _slug(desc.get("parameter_name") or pcode)
@@ -257,6 +272,16 @@ def _build_timeseries(
             name += "_x"
         used.add(name)
 
+        if len(group_units) > 1:
+            # One variable can carry only one ``units`` attr; surface the
+            # mix instead of silently labeling every value with the first.
+            _warnings.warn(
+                f"'{name}' spans multiple units {list(group_units)}; labeling "
+                f"with '{unit}'. Filter by site/parameter to avoid mixing "
+                "units in one variable.",
+                stacklevel=3,
+            )
+
         sub = group.set_index([_INSTANCE, "time"])[["value", *ancillary]]
         if not sub.index.is_unique:
             _warnings.warn(

tests/waterdata_xarray_test.py

@@ -216,3 +216,42 @@ def spy(func, args, kwargs):
         # the wrapper hands include_hash to _fetch; _fetch is what drops it
         # (asserted in test_fetch_strips_include_hash)
         assert seen[0].get("include_hash") is True
+
+
+def test_unparseable_time_dropped_with_warning():
+    # A bad/missing time must be dropped explicitly (with a warning), not
+    # silently swallowed by to_xarray.
+    df = _daily_frame(
+        values=(100, 110, 120),
+        times=("2024-06-01", "not-a-date", "2024-06-03"),
+    )
+    with pytest.warns(UserWarning, match="unparseable or missing time"):
+        ds = wdx._build_timeseries(
+            df, _meta(), service="daily", series_meta=_DISCHARGE_META
+        )
+    assert ds.sizes["time"] == 2  # the bad-time row is gone, the good ones stay
+    assert 110 not in ds["discharge"].values  # the dropped value did not survive
+
+
+def test_all_unparseable_time_returns_empty_dataset():
+    df = _daily_frame(values=(1, 2), times=("bad-a", "bad-b"))
+    with pytest.warns(UserWarning, match="unparseable or missing time"):
+        ds = wdx._build_timeseries(
+            df, _meta(), service="daily", series_meta=_DISCHARGE_META
+        )
+    assert list(ds.data_vars) == []
+    assert ds.attrs["Conventions"] == "CF-1.11"
+
+
+def test_mixed_units_in_one_variable_warns():
+    # Same (parameter, statistic) across two sites but different units -> one
+    # variable can hold only one units attr; warn instead of silently mislabeling.
+    a = _daily_frame(site="USGS-1", values=(100,), times=("2024-06-01",))
+    b = _daily_frame(site="USGS-2", values=(3,), times=("2024-06-01",))
+    b["unit_of_measure"] = "m3 s-1"
+    with pytest.warns(UserWarning, match="spans multiple units"):
+        ds = wdx._build_timeseries(
+            pd.concat([a, b]), _meta(), service="daily", series_meta=_DISCHARGE_META
+        )
+    assert ds.sizes["monitoring_location_id"] == 2
+    assert ds["discharge"].attrs["units"] == "ft3 s-1"  # first unit (ft^3/s)