Add major-chemistry data product (one feature per well)

backend/persisters/ogc_features.py

@@ -77,6 +77,103 @@ def dump_summary_collection(path: str, records: list, meta: dict) -> dict:
     return collection
 
 
+def dump_major_chemistry_collection(path: str, records: list, meta: dict) -> dict:
+    """
+    Write an OGC FeatureCollection of wells with major-chemistry analytes as
+    properties to *path*. One Feature per well.
+
+    *records* is a flat list of SummaryRecord — one per (well, analyte). They are
+    pivoted by well: each analyte contributes ``<analyte>`` (latest value),
+    ``<analyte>_units``, and ``<analyte>_date`` properties. Well identity is
+    ``(source, id)``; well_depth and geometry come from any of the well's
+    records.
+
+    Returns the collection dict (for testing).
+    §V: MUST include top-level id, type, numberReturned, timeStamp.
+    §V: Each Feature MUST have top-level id.
+    """
+    collection_id = meta.get("id", "collection")
+
+    wells: dict = {}
+    for r in records:
+        source = getattr(r, "source", "") or ""
+        rid = getattr(r, "id", "") or ""
+        key = (source, rid)
+        well = wells.get(key)
+        if well is None:
+            well = {
+                "source": source,
+                "id": rid,
+                "name": getattr(r, "name", None),
+                "latitude": getattr(r, "latitude", None),
+                "longitude": getattr(r, "longitude", None),
+                "elevation": getattr(r, "elevation", None),
+                "well_depth": getattr(r, "well_depth", None),
+                "well_depth_units": getattr(r, "well_depth_units", None),
+                "analytes": {},
+            }
+            wells[key] = well
+        # well_depth can be absent on some analyte records; keep first non-null.
+        if well["well_depth"] is None and getattr(r, "well_depth", None) is not None:
+            well["well_depth"] = getattr(r, "well_depth", None)
+            well["well_depth_units"] = getattr(r, "well_depth_units", None)
+
+        analyte = getattr(r, "parameter_name", None)
+        if analyte:
+            well["analytes"][analyte] = {
+                "value": getattr(r, "latest_value", None),
+                "units": getattr(r, "latest_units", None),
+                "date": getattr(r, "latest_date", None),
+            }
+
+    features = []
+    for (source, rid), well in wells.items():
+        feature_id = f"{source}:{rid}" if source and rid else str(rid)
+        props = {
+            "source": source,
+            "id": rid,
+            "name": well["name"],
+            "well_depth": well["well_depth"],
+            "well_depth_units": well["well_depth_units"],
+        }
+        for analyte, vals in well["analytes"].items():
+            props[analyte] = vals["value"]
+            props[f"{analyte}_units"] = vals["units"]
+            props[f"{analyte}_date"] = vals["date"]
+
+        lat, lon, elev = well["latitude"], well["longitude"], well["elevation"]
+        coords = [lon, lat] if elev is None else [lon, lat, elev]
+        features.append({
+            "type": "Feature",
+            "id": feature_id,
+            "geometry": {"type": "Point", "coordinates": coords},
+            "properties": props,
+        })
+
+    collection = {
+        "type": "FeatureCollection",
+        "id": collection_id,
+        "title": meta.get("title", collection_id),
+        "description": meta.get("description", ""),
+        "timeStamp": _timestamp_now(),
+        "numberMatched": len(features),
+        "numberReturned": len(features),
+        "links": [
+            {
+                "href": meta.get("href", ""),
+                "rel": "self",
+                "type": "application/geo+json",
+            }
+        ],
+        "features": features,
+    }
+
+    with open(path, "w", encoding="utf-8") as f:
+        json.dump(collection, f, indent=2, default=str)
+
+    return collection
+
+
 def dump_timeseries_collection(
     path: str,
     site_records: list,

orchestration/assets/products.py

@@ -34,6 +34,7 @@
 
 from backend.config import PARAMETER_SOURCE_MAP, WATERLEVELS
 from backend.persisters.ogc_features import (
+    dump_major_chemistry_collection,
     dump_summary_collection,
     dump_timeseries_collection,
 )
@@ -47,17 +48,42 @@
 _CHECK_NAME = "returned_data"
 _GEOSERVER_CHECK_NAME = "registered"
 
+# Classic major-ion suite for the ogc_major_chemistry product. One feature per
+# well, with each analyte's latest value/units/date as properties.
+_MAJOR_CHEMISTRY = [
+    "calcium",
+    "magnesium",
+    "sodium",
+    "potassium",
+    "bicarbonate",
+    "carbonate",
+    "chloride",
+    "sulfate",
+]
+
+
+def _product_params(product: dict) -> list:
+    """The DIE parameter(s) a product unifies. Single-parameter products yield
+    one; the major-chemistry product yields the major-ion suite."""
+    if product.get("output_type") == "ogc_major_chemistry":
+        return list(_MAJOR_CHEMISTRY)
+    return [product["parameter"]]
+
 
 def _product_source_keys(product: dict) -> list:
-    """Source keys that apply to this product: the parameter's agencies,
-    filtered by the product's include/exclude list."""
-    agencies = PARAMETER_SOURCE_MAP[product["parameter"]]["agencies"]
+    """Source keys that apply to this product: the union of its parameters'
+    agencies, filtered by the product's include/exclude list."""
+    agencies: list = []
+    for param in _product_params(product):
+        for a in PARAMETER_SOURCE_MAP[param]["agencies"]:
+            if a not in agencies:
+                agencies.append(a)
     spec = product.get("sources", {}) or {}
     if spec.get("include"):
         return [a for a in agencies if a in spec["include"]]
     if spec.get("exclude"):
         return [a for a in agencies if a not in spec["exclude"]]
-    return list(agencies)
+    return agencies
 
 
 def _in_name(source_key: str) -> str:
@@ -75,24 +101,32 @@ def _build_source_asset(product: dict, source_key: str, group: str):
     plain ``_payload`` dicts for IO-manager pickling (see module docstring)."""
     pid = product["id"]
     src_key = dg.AssetKey([pid, "sources", source_key])
+    params = _product_params(product)
 
     @dg.asset(
         key=src_key,
         group_name=group,
         check_specs=[dg.AssetCheckSpec(name=_CHECK_NAME, asset=src_key)],
     )
     def _source_asset(context: dg.AssetExecutionContext, die_config: DIEConfigResource):
-        config = die_config.get_config(product)
-
         error = ""
         records, sites, timeseries = [], [], []
         try:
+            # One unification pass per parameter. Single-parameter products run
+            # once; the major-chemistry product runs once per analyte and
+            # accumulates summary records (analyte identity lives in each
+            # record's parameter_name). A source that doesn't provide a given
+            # parameter is skipped by unify_source (source_pair → None).
             with forward_die_logs(context):
-                persister = unify_source(config, source_key)
-            # Ship plain dicts across the IO manager; rebuild in combine.
-            records = [r._payload for r in persister.records]
-            sites = [s._payload for s in persister.sites]
-            timeseries = [[o._payload for o in site_ts] for site_ts in persister.timeseries]
+                for param in params:
+                    config = die_config.get_config(product, parameter=param)
+                    persister = unify_source(config, source_key)
+                    # Ship plain dicts across the IO manager; rebuild in combine.
+                    records.extend(r._payload for r in persister.records)
+                    sites.extend(s._payload for s in persister.sites)
+                    timeseries.extend(
+                        [o._payload for o in site_ts] for site_ts in persister.timeseries
+                    )
         except Exception:
             error = traceback.format_exc()
             context.log.error(f"Source {source_key} failed:\n{error}")
@@ -132,10 +166,11 @@ def _build_combine_asset(product: dict, source_keys: list, source_asset_keys: li
     """Build the combine asset (keyed ``[product_id]``) for *product*.
 
     Depends on every source asset (wired via ``ins``), merges their
-    records/sites/timeseries, writes the OGC GeoJSON collection — summary or
-    timeseries depending on ``output_type`` — and uploads it to GCS."""
+    records/sites/timeseries, writes the OGC GeoJSON collection — summary,
+    timeseries, or major-chemistry depending on ``output_type`` — and uploads it
+    to GCS."""
     pid = product["id"]
-    is_summary = product["output_type"] == "ogc_summary"
+    output_type = product["output_type"]
     ins = {
         _in_name(k): dg.AssetIn(key=ak)
         for k, ak in zip(source_keys, source_asset_keys)
@@ -161,7 +196,12 @@ def _combine_asset(
 
         with tempfile.TemporaryDirectory() as tmpdir:
             out = Path(tmpdir) / "collection.geojson"
-            if is_summary:
+            if output_type == "ogc_major_chemistry":
+                # All summary records (one per well+analyte); the dumper pivots
+                # to one feature per well with analytes as properties.
+                records = [SummaryRecord(p) for p in all_records]
+                dump_major_chemistry_collection(str(out), records, meta)
+            elif output_type == "ogc_summary":
                 records = [SummaryRecord(p) for p in all_records]
                 dump_summary_collection(str(out), records, meta)
             else:
@@ -275,8 +315,9 @@ def build_product_assets(product: dict) -> list:
     """Return the full asset list for *product*: one source asset per applicable
     source, the combine asset, and the geoserver publish asset (see module
     docstring for the graph shape). Assets are grouped ``waterlevels`` or
-    ``analytes`` by parameter."""
-    group = "waterlevels" if product["parameter"] == WATERLEVELS else "analytes"
+    ``analytes`` by parameter (major-chemistry products group under
+    ``analytes``)."""
+    group = "waterlevels" if product.get("parameter") == WATERLEVELS else "analytes"
     source_keys = _product_source_keys(product)
 
     source_assets = []

orchestration/config/products.yaml

@@ -55,3 +55,16 @@ products:
       state: NM
     sources:
       exclude: []
+
+  # One feature per well; each major-ion analyte's latest value/units/date,
+  # plus well depth, stored as properties. Analyte set is fixed in
+  # assets/products.py (_MAJOR_CHEMISTRY), so no `parameter` here.
+  - id: nm_major_chemistry
+    output_type: ogc_major_chemistry
+    title: "NM Major Chemistry"
+    description: "Wells with major-ion chemistry (latest value per analyte), well depth, all NM sources"
+    schedule: "0 11 * * *"
+    spatial_filter:
+      state: NM
+    sources:
+      exclude: []

orchestration/definitions.py

@@ -11,7 +11,7 @@
 
 _PRODUCTS_PATH = Path(__file__).parent / "config" / "products.yaml"
 
-_SUPPORTED_OUTPUT_TYPES = {"ogc_summary", "ogc_timeseries"}
+_SUPPORTED_OUTPUT_TYPES = {"ogc_summary", "ogc_timeseries", "ogc_major_chemistry"}
 
 
 def _load_products() -> dict:

orchestration/resources/die_config.py

@@ -8,25 +8,36 @@ class DIEConfigResource(dg.ConfigurableResource):
 
     usgs_api_key: Optional[str] = None
 
-    def get_config(self, product: dict) -> Config:
+    def get_config(self, product: dict, parameter: Optional[str] = None) -> Config:
         """Translate a products.yaml entry into a finalized DIE ``Config``.
 
         Mapping:
-        - ``output_type`` → ``output_summary`` / ``output_format``.
+        - ``output_type`` → ``output_summary`` / ``output_format``. Both
+          ``ogc_summary`` and ``ogc_major_chemistry`` run in summary mode (the
+          latter pivots per-analyte summaries into one feature per well).
         - ``spatial_filter.county`` → ``county``. ``spatial_filter.state`` sets
           ``wkt = None`` (statewide; DIE applies the NM extent downstream).
         - ``sources.include`` → enable only those sources (all others off).
           ``sources.exclude`` → disable those, leave the rest at their defaults.
         - ``parameter`` is set on the Config, then ``finalize()`` validates and
           resolves output units/paths.
+
+        *parameter* overrides ``product["parameter"]`` — used by the
+        major-chemistry product, which has no single parameter and calls this
+        once per analyte.
         """
         spatial = product.get("spatial_filter", {})
         sources_spec = product.get("sources", {})
 
+        output_type = product.get("output_type", "ogc_summary")
+        is_summary = output_type in ("ogc_summary", "ogc_major_chemistry")
+
         payload: dict = {
             "yes": True,
-            "output_summary": product.get("output_type") == "ogc_summary",
-            "output_format": product.get("output_type", "ogc_summary"),
+            "output_summary": is_summary,
+            # backend only distinguishes summary vs timeseries; major-chemistry
+            # is a summary variant as far as unification is concerned.
+            "output_format": "ogc_summary" if is_summary else output_type,
         }
 
         if spatial.get("county"):
@@ -49,6 +60,6 @@ def get_config(self, product: dict) -> Config:
                 payload[f"use_source_{s}"] = False
 
         config = Config(payload=payload)
-        config.parameter = product["parameter"]
+        config.parameter = parameter or product["parameter"]
         config.finalize()
         return config

tests/test_persisters/test_ogc_features.py

@@ -2,7 +2,11 @@
 import os
 import tempfile
 
-from backend.persisters.ogc_features import dump_summary_collection, dump_timeseries_collection
+from backend.persisters.ogc_features import (
+    dump_summary_collection,
+    dump_timeseries_collection,
+    dump_major_chemistry_collection,
+)
 from backend.record import SummaryRecord, SiteRecord, ParameterRecord
 
 
@@ -172,3 +176,55 @@ def test_ogc_required_fields(self, tmp_path):
         assert result["id"] == "nm_ts"
         assert "timeStamp" in result
         assert "numberReturned" in result
+
+
+def _make_chem_record(source, rid, analyte, value, units="mg/L", date="2024-05-01", well_depth=None):
+    return SummaryRecord({
+        "source": source,
+        "id": rid,
+        "name": f"Well {rid}",
+        "latitude": 34.0,
+        "longitude": -106.0,
+        "elevation": None,
+        "well_depth": well_depth,
+        "well_depth_units": "ft",
+        "parameter_name": analyte,
+        "latest_value": value,
+        "latest_units": units,
+        "latest_date": date,
+    })
+
+
+class TestMajorChemistryCollection:
+    def test_pivots_analytes_into_one_feature_per_well(self, tmp_path):
+        records = [
+            _make_chem_record("NMBGMR", "W1", "calcium", 42.0, well_depth=120.0),
+            _make_chem_record("NMBGMR", "W1", "chloride", 15.0),
+            _make_chem_record("WQP", "W2", "calcium", 55.0),
+        ]
+        out = tmp_path / "mc.geojson"
+        result = dump_major_chemistry_collection(str(out), records, {"id": "nm_major_chemistry"})
+
+        assert result["numberReturned"] == 2  # two distinct wells
+        by_id = {f["id"]: f for f in result["features"]}
+
+        w1 = by_id["NMBGMR:W1"]["properties"]
+        assert w1["calcium"] == 42.0
+        assert w1["calcium_units"] == "mg/L"
+        assert w1["calcium_date"] == "2024-05-01"
+        assert w1["chloride"] == 15.0
+        assert w1["well_depth"] == 120.0  # carried from the record that had it
+
+        w2 = by_id["WQP:W2"]["properties"]
+        assert w2["calcium"] == 55.0
+        assert "chloride" not in w2  # missing analyte omitted
+
+    def test_geometry_and_required_fields(self, tmp_path):
+        out = tmp_path / "mc.geojson"
+        result = dump_major_chemistry_collection(
+            str(out), [_make_chem_record("NMBGMR", "W1", "sodium", 30.0)], {"id": "nm_major_chemistry"}
+        )
+        assert result["type"] == "FeatureCollection"
+        assert "timeStamp" in result
+        feat = result["features"][0]
+        assert feat["geometry"]["coordinates"] == [-106.0, 34.0]