Add Ocotillo OGC API source (summary-only) alongside AMP

backend/config.py

@@ -26,6 +26,11 @@
     NMBGMRWaterLevelSource,
     NMBGMRAnalyteSource,
 )
+from .connectors.ocotillo.source import (
+    OcotilloSiteSource,
+    OcotilloWaterLevelSource,
+    OcotilloAnalyteSource,
+)
 from .connectors.bor.source import BORSiteSource, BORAnalyteSource
 from .connectors.nmenv.source import DWBSiteSource, DWBAnalyteSource
 from .connectors.nmose.source import NMOSEPODSiteSource
@@ -76,24 +81,24 @@
 
 
 PARAMETER_SOURCE_MAP = {
-    WATERLEVELS: {"agencies": ["bernco", "cabq", "ebid", "nmbgmr_amp", "nmose_isc_seven_rivers", "nmose_roswell", "nwis", "pvacd", "wqp"]},
-    CARBONATE: {"agencies": ["nmbgmr_amp", "wqp"]},
-    ARSENIC: {"agencies": ["bor", "nmbgmr_amp", "nmed_dwb", "wqp"]},
-    URANIUM: {"agencies": ["bor", "nmbgmr_amp", "nmed_dwb", "wqp"]},
-    SPECIFIC_CONDUCTANCE: {"agencies": ["nmbgmr_amp", "nmed_dwb", "nmose_isc_seven_rivers", "wqp"]},
+    WATERLEVELS: {"agencies": ["bernco", "cabq", "ebid", "nmbgmr_amp", "nmose_isc_seven_rivers", "nmose_roswell", "nwis", "ocotillo", "pvacd", "wqp"]},
+    CARBONATE: {"agencies": ["nmbgmr_amp", "ocotillo", "wqp"]},
+    ARSENIC: {"agencies": ["bor", "nmbgmr_amp", "nmed_dwb", "ocotillo", "wqp"]},
+    URANIUM: {"agencies": ["bor", "nmbgmr_amp", "nmed_dwb", "ocotillo", "wqp"]},
+    SPECIFIC_CONDUCTANCE: {"agencies": ["nmbgmr_amp", "nmed_dwb", "nmose_isc_seven_rivers", "ocotillo", "wqp"]},
     CONDUCTIVITY: {"agencies": ["bor", "nmose_isc_seven_rivers", "wqp"]},
-    BICARBONATE: {"agencies": ["nmbgmr_amp", "nmed_dwb", "nmose_isc_seven_rivers", "wqp"]},
-    CALCIUM: {"agencies": ["bor", "nmbgmr_amp", "nmed_dwb", "nmose_isc_seven_rivers", "wqp"]},
-    CHLORIDE: {"agencies": ["bor", "nmbgmr_amp", "nmed_dwb", "nmose_isc_seven_rivers", "wqp"]},
-    FLUORIDE: {"agencies": ["bor", "nmbgmr_amp", "nmed_dwb", "nmose_isc_seven_rivers", "wqp"]},
-    MAGNESIUM: {"agencies": ["bor", "nmbgmr_amp", "nmed_dwb", "nmose_isc_seven_rivers", "wqp"]},
-    NITRATE: {"agencies": ["bor", "nmbgmr_amp", "nmed_dwb", "nmose_isc_seven_rivers", "wqp"]},
-    PH: {"agencies": ["bor", "nmbgmr_amp", "nmed_dwb", "nmose_isc_seven_rivers", "wqp"]},
-    POTASSIUM: {"agencies": ["bor", "nmbgmr_amp", "nmed_dwb", "nmose_isc_seven_rivers", "wqp"]},
-    SILICA: {"agencies": ["bor", "nmbgmr_amp", "nmed_dwb", "nmose_isc_seven_rivers", "wqp"]},
-    SODIUM: {"agencies": ["bor", "nmbgmr_amp", "nmed_dwb", "nmose_isc_seven_rivers", "wqp"]},
-    SULFATE: {"agencies": ["bor", "nmbgmr_amp", "nmed_dwb", "nmose_isc_seven_rivers", "wqp"]},
-    TDS: {"agencies": ["bor", "nmbgmr_amp", "nmed_dwb", "nmose_isc_seven_rivers", "wqp"]},
+    BICARBONATE: {"agencies": ["nmbgmr_amp", "nmed_dwb", "nmose_isc_seven_rivers", "ocotillo", "wqp"]},
+    CALCIUM: {"agencies": ["bor", "nmbgmr_amp", "nmed_dwb", "nmose_isc_seven_rivers", "ocotillo", "wqp"]},
+    CHLORIDE: {"agencies": ["bor", "nmbgmr_amp", "nmed_dwb", "nmose_isc_seven_rivers", "ocotillo", "wqp"]},
+    FLUORIDE: {"agencies": ["bor", "nmbgmr_amp", "nmed_dwb", "nmose_isc_seven_rivers", "ocotillo", "wqp"]},
+    MAGNESIUM: {"agencies": ["bor", "nmbgmr_amp", "nmed_dwb", "nmose_isc_seven_rivers", "ocotillo", "wqp"]},
+    NITRATE: {"agencies": ["bor", "nmbgmr_amp", "nmed_dwb", "nmose_isc_seven_rivers", "ocotillo", "wqp"]},
+    PH: {"agencies": ["bor", "nmbgmr_amp", "nmed_dwb", "nmose_isc_seven_rivers", "ocotillo", "wqp"]},
+    POTASSIUM: {"agencies": ["bor", "nmbgmr_amp", "nmed_dwb", "nmose_isc_seven_rivers", "ocotillo", "wqp"]},
+    SILICA: {"agencies": ["bor", "nmbgmr_amp", "nmed_dwb", "nmose_isc_seven_rivers", "ocotillo", "wqp"]},
+    SODIUM: {"agencies": ["bor", "nmbgmr_amp", "nmed_dwb", "nmose_isc_seven_rivers", "ocotillo", "wqp"]},
+    SULFATE: {"agencies": ["bor", "nmbgmr_amp", "nmed_dwb", "nmose_isc_seven_rivers", "ocotillo", "wqp"]},
+    TDS: {"agencies": ["bor", "nmbgmr_amp", "nmed_dwb", "nmose_isc_seven_rivers", "ocotillo", "wqp"]},
 }
 
 SOURCE_DICT = {
@@ -107,6 +112,7 @@
     "nmose_pod": NMOSEPODSiteSource,
     "nmose_roswell": NMOSERoswellSiteSource,
     "nwis": NWISSiteSource,
+    "ocotillo": OcotilloSiteSource,
     "pvacd": PVACDSiteSource,
     "wqp": WQPSiteSource,
 }
@@ -123,6 +129,7 @@
     "nmose_isc_seven_rivers": (ISCSevenRiversSiteSource, ISCSevenRiversAnalyteSource),
     "nmbgmr_amp": (NMBGMRSiteSource, NMBGMRAnalyteSource),
     "nmed_dwb": (DWBSiteSource, DWBAnalyteSource),
+    "ocotillo": (OcotilloSiteSource, OcotilloAnalyteSource),
 }
 
 WATERLEVEL_SOURCE_PAIRS = {
@@ -134,6 +141,7 @@
     "bernco": (BernCoSiteSource, BernCoWaterLevelSource),
     "ebid": (EBIDSiteSource, EBIDWaterLevelSource),
     "cabq": (CABQSiteSource, CABQWaterLevelSource),
+    "ocotillo": (OcotilloSiteSource, OcotilloWaterLevelSource),
     "wqp": (WQPSiteSource, WQPWaterLevelSource),
 }
 
@@ -180,6 +188,7 @@ class Config:
     use_source_nmose_pod: bool = True
     use_source_nmose_roswell: bool = True
     use_source_nwis: bool = True
+    use_source_ocotillo: bool = True
     use_source_pvacd: bool = True
     use_source_wqp: bool = True
 

backend/connectors/ocotillo/mappings.py

@@ -0,0 +1,79 @@
+# ===============================================================================
+# Copyright 2024 Jake Ross
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ===============================================================================
+"""
+Ocotillo OGC API - Features (pygeoapi/PostGIS) connector mappings.
+
+The Ocotillo API (ocotillo-api.newmexicowaterdata.org/ogcapi) is intended as a
+replacement for the NMBGMR AMP API. Unlike AMP, it does NOT expose raw
+per-observation time series. It only publishes "latest" / "summary" snapshot
+collections. Therefore this connector supports SUMMARY output only; time series
+output is unsupported (see OcotilloWaterLevelSource/OcotilloAnalyteSource).
+
+Because the API pre-aggregates, several DIG summary columns cannot be populated
+and are intentionally left null (per project decision): mean, earliest_* for all
+parameters; nrecords/min/max for chemistry and TDS (only water levels expose
+count/min/max via water_well_summary).
+"""
+from backend.constants import (
+    ARSENIC,
+    BICARBONATE,
+    CALCIUM,
+    CARBONATE,
+    CHLORIDE,
+    FLUORIDE,
+    MAGNESIUM,
+    NITRATE,
+    PH,
+    POTASSIUM,
+    SILICA,
+    SODIUM,
+    SPECIFIC_CONDUCTANCE,
+    SULFATE,
+    TDS,
+    URANIUM,
+)
+
+# Collection ids on the Ocotillo OGC API
+SITE_COLLECTION = "water_wells"
+WATERLEVEL_SUMMARY_COLLECTION = "water_well_summary"
+MAJOR_CHEMISTRY_COLLECTION = "major_chemistry_results"
+MINOR_CHEMISTRY_COLLECTION = "minor_chemistry_wells"
+TDS_COLLECTION = "latest_tds_wells"
+
+# DIG analyte -> (Ocotillo collection, property/column holding the latest value).
+# Major-ion / field chemistry live in major_chemistry_results as static analyte
+# columns; trace metals live in minor_chemistry_wells; TDS has its own
+# latest_tds_wells collection with bespoke field names (handled in the source).
+OCOTILLO_ANALYTE_MAPPING: dict = {
+    CALCIUM: (MAJOR_CHEMISTRY_COLLECTION, "calcium"),
+    MAGNESIUM: (MAJOR_CHEMISTRY_COLLECTION, "magnesium"),
+    SODIUM: (MAJOR_CHEMISTRY_COLLECTION, "sodium"),
+    POTASSIUM: (MAJOR_CHEMISTRY_COLLECTION, "potassium"),
+    BICARBONATE: (MAJOR_CHEMISTRY_COLLECTION, "bicarbonate"),
+    CARBONATE: (MAJOR_CHEMISTRY_COLLECTION, "carbonate"),
+    SULFATE: (MAJOR_CHEMISTRY_COLLECTION, "sulfate"),
+    CHLORIDE: (MAJOR_CHEMISTRY_COLLECTION, "chloride"),
+    NITRATE: (MAJOR_CHEMISTRY_COLLECTION, "nitrate"),
+    FLUORIDE: (MAJOR_CHEMISTRY_COLLECTION, "fluoride"),
+    SILICA: (MAJOR_CHEMISTRY_COLLECTION, "silica"),
+    PH: (MAJOR_CHEMISTRY_COLLECTION, "ph"),
+    SPECIFIC_CONDUCTANCE: (MAJOR_CHEMISTRY_COLLECTION, "specific_conductance"),
+    ARSENIC: (MINOR_CHEMISTRY_COLLECTION, "arsenic"),
+    URANIUM: (MINOR_CHEMISTRY_COLLECTION, "uranium"),
+    TDS: (TDS_COLLECTION, "latest_tds_value"),
+}
+
+# ============= EOF =============================================

backend/connectors/ocotillo/source.py

@@ -0,0 +1,178 @@
+# ===============================================================================
+# Copyright 2024 Jake Ross
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ===============================================================================
+"""
+Ocotillo OGC API - Features connector.
+
+Ocotillo (ocotillo-api.newmexicowaterdata.org/ogcapi) is a pygeoapi/PostGIS
+service intended to replace the NMBGMR AMP API. It publishes only pre-aggregated
+"latest"/"summary" feature collections and exposes NO raw observation time
+series. Consequently this connector produces SUMMARY output only; requesting
+time series output logs a warning and yields nothing.
+
+Sites are seeded from the ``water_wells`` collection. Water-level summaries come
+from ``water_well_summary`` (count/min/max/latest); chemistry and TDS come from
+their respective latest-value collections. See mappings.py for details and for
+the list of summary columns that are intentionally null.
+"""
+import os
+
+from backend.connectors import NM_STATE_BOUNDING_POLYGON
+from backend.connectors.ocotillo.mappings import (
+    OCOTILLO_ANALYTE_MAPPING,
+    SITE_COLLECTION,
+    WATERLEVEL_SUMMARY_COLLECTION,
+)
+from backend.connectors.ocotillo.transformer import (
+    OcotilloSiteTransformer,
+    OcotilloWaterLevelTransformer,
+    OcotilloAnalyteTransformer,
+)
+from backend.source import (
+    BaseSiteSource,
+    BaseWaterLevelSource,
+    BaseAnalyteSource,
+)
+
+TIMEOUT = 15 * 60
+DEFAULT_URL = "https://ocotillo-api.newmexicowaterdata.org/ogcapi"
+
+
+def _base_url():
+    return os.getenv("OCOTILLO_URL", DEFAULT_URL).rstrip("/")
+
+
+def _bbox_params(config):
+    """OGC API bbox filter (minx,miny,maxx,maxy) from the configured bounds.
+    Rectangular only; finer WKT/county filtering happens in the transformer's
+    geographic filter."""
+    params = {}
+    if config.has_bounds():
+        x1, y1, x2, y2 = config.bbox_bounding_points()
+        params["bbox"] = f"{x1},{y1},{x2},{y2}"
+    return params
+
+
+def _fetch_all_features(source, collection, params):
+    """Page through an OGC API - Features collection, returning all features."""
+    url = f"{_base_url()}/collections/{collection}/items"
+    limit = 1000
+    offset = 0
+    features = []
+    while True:
+        page_params = {"f": "json", "limit": limit, "offset": offset}
+        page_params.update(params)
+        fc = source._execute_json_request(url, page_params, timeout=TIMEOUT)
+        page = fc.get("features", []) if isinstance(fc, dict) else []
+        features.extend(page)
+        if len(page) < limit:
+            break
+        offset += limit
+    return features
+
+
+class OcotilloSiteSource(BaseSiteSource):
+    chunk_size = 100
+    bounding_polygon = NM_STATE_BOUNDING_POLYGON
+
+    def __init__(self):
+        super().__init__(transformer=OcotilloSiteTransformer())
+
+    def __repr__(self):
+        return "OcotilloSiteSource"
+
+    def health(self):
+        try:
+            url = f"{_base_url()}/collections/{SITE_COLLECTION}/items"
+            resp = self._execute_json_request(url, {"f": "json", "limit": 1})
+            return bool(resp)
+        except Exception:
+            return False
+
+    def get_records(self):
+        return _fetch_all_features(self, SITE_COLLECTION, _bbox_params(self.config))
+
+
+class _OcotilloSummaryParameterSource:
+    """Shared behavior for Ocotillo parameter sources.
+
+    The whole (bbox-filtered) collection is fetched once and cached keyed by
+    well name; per-site extraction is a dict lookup. Summarization is a direct
+    passthrough to the transformer because the API already aggregated the data.
+    Time series output is unsupported.
+    """
+
+    def _get_collection(self):
+        raise NotImplementedError
+
+    def get_records(self, site_record):
+        cache = getattr(self, "_feature_cache", None)
+        if cache is None:
+            cache = {}
+            for feature in _fetch_all_features(
+                self, self._get_collection(), _bbox_params(self.config)
+            ):
+                name = feature.get("properties", {}).get("name")
+                if name is not None:
+                    cache[name] = feature
+            self._feature_cache = cache
+        return cache
+
+    def _extract_site_records(self, records, site_record):
+        feature = records.get(site_record.id)
+        return [feature] if feature is not None else []
+
+    def _summarize_records(self, site, cleaned):
+        # cleaned is the single pre-aggregated feature for this site.
+        return self.transformer.do_transform(cleaned[0], site)
+
+    def read_timeseries(self, site_record):
+        self.warn(
+            "Ocotillo source supports summary output only; the API exposes no "
+            "raw time series. Skipping timeseries output."
+        )
+        return None
+
+
+class OcotilloWaterLevelSource(_OcotilloSummaryParameterSource, BaseWaterLevelSource):
+    def __init__(self):
+        super().__init__(transformer=OcotilloWaterLevelTransformer())
+
+    def __repr__(self):
+        return "OcotilloWaterLevelSource"
+
+    def _get_collection(self):
+        return WATERLEVEL_SUMMARY_COLLECTION
+
+
+class OcotilloAnalyteSource(_OcotilloSummaryParameterSource, BaseAnalyteSource):
+    def __init__(self):
+        super().__init__(transformer=OcotilloAnalyteTransformer())
+
+    def __repr__(self):
+        return "OcotilloAnalyteSource"
+
+    def _get_collection(self):
+        try:
+            collection, _column = OCOTILLO_ANALYTE_MAPPING[self.config.parameter]
+        except KeyError:
+            raise ValueError(
+                f"Ocotillo source does not provide parameter "
+                f"{self.config.parameter!r}. Valid: {sorted(OCOTILLO_ANALYTE_MAPPING)}"
+            )
+        return collection
+
+
+# ============= EOF =============================================