Dedup NWIS site time-series features by location

backend/connectors/usgs/source.py

@@ -186,7 +186,28 @@ def get_records(self):
         records: list = data.get("features", [])
         self._requester.check_truncation(data, "site")
 
-        return records
+        # combined-metadata returns one feature per time series (data_type /
+        # statistic), so a location with multiple series (e.g. field
+        # measurements + daily mean/max/min) appears several times with the
+        # same monitoring_location_id and identical site metadata. Left as-is,
+        # read_timeseries iterates each duplicate site and re-emits that well's
+        # readings once per series, producing exact-duplicate observations
+        # downstream. Keep the first feature per location; readings come only
+        # from the field-measurements collection regardless of series.
+        deduped: list = []
+        seen: set = set()
+        for feature in records:
+            site_id = feature.get("properties", {}).get("monitoring_location_id")
+            if site_id in seen:
+                continue
+            seen.add(site_id)
+            deduped.append(feature)
+
+        removed = len(records) - len(deduped)
+        if removed:
+            self.warn(f"Dropped {removed} duplicate site time-series features ({len(deduped)} unique locations)")
+
+        return deduped
 
 
 class NWISWaterLevelSource(BaseWaterLevelSource):

tests/test_sources/test_nwis.py

@@ -23,3 +23,39 @@ class TestNWISWaterlevels(BaseSourceTestClass):
     parameter = WATERLEVELS
     units = FEET
     agency = "nwis"
+
+
+def _site_feature(loc_id, data_type):
+    return {
+        "properties": {
+            "monitoring_location_id": loc_id,
+            "data_type": data_type,
+            "monitoring_location_name": "TOME SITE",
+        },
+        "geometry": {"type": "Point", "coordinates": [-106.6, 34.7]},
+    }
+
+
+def test_nwis_site_source_dedups_duplicate_timeseries_features():
+    # combined-metadata returns one feature per time series, so a well with
+    # several series (field measurements + daily mean/max/min) shows up multiple
+    # times with the same monitoring_location_id. get_records must collapse these
+    # to one site so its readings aren't re-emitted once per series downstream.
+    from backend.config import Config
+    from backend.connectors.usgs.source import NWISSiteSource
+
+    features = [
+        _site_feature("USGS-344431106393403", "Field measurements"),
+        _site_feature("USGS-344431106393403", "Daily values"),
+        _site_feature("USGS-344431106393403", "Continuous values"),
+        _site_feature("USGS-999900000000000", "Field measurements"),
+    ]
+
+    source = NWISSiteSource()
+    source.set_config(Config())
+    source._requester.request = lambda *a, **k: {"features": features, "links": []}
+
+    records = source.get_records()
+
+    ids = [f["properties"]["monitoring_location_id"] for f in records]
+    assert ids == ["USGS-344431106393403", "USGS-999900000000000"]