From b7c7d4cc178a87d46a501e4dd68820ecaea4efdf Mon Sep 17 00:00:00 2001 From: jakeross Date: Fri, 26 Jun 2026 09:16:37 -0600 Subject: [PATCH] Add major-chemistry data product (one feature per well) New ogc_major_chemistry product: nm_major_chemistry. One GeoJSON feature per well, with each major-ion analyte's latest value/units/date plus well depth as properties. - backend/persisters/ogc_features.py: dump_major_chemistry_collection pivots per-(well,analyte) SummaryRecords into one feature per well, keyed (source, id); carries well_depth and geometry. - orchestration/assets/products.py: _MAJOR_CHEMISTRY (classic 8 major ions); products now unify over a list of parameters (single-parameter products run once, major-chemistry runs once per analyte and accumulates); source keys = union of the analytes' agencies; combine picks the dumper by output_type. - die_config.get_config: optional parameter override; treat ogc_major_chemistry as summary mode. - definitions.py: register ogc_major_chemistry as a supported output type (gets its own per-product job + schedule automatically). - products.yaml: nm_major_chemistry entry (all NM sources). - tests: pivot + geometry/required-field coverage for the new dumper. Co-Authored-By: Claude Opus 4.8 --- backend/persisters/ogc_features.py | 97 ++++++++++++++++++++++ orchestration/assets/products.py | 75 +++++++++++++---- orchestration/config/products.yaml | 13 +++ orchestration/definitions.py | 2 +- orchestration/resources/die_config.py | 21 +++-- tests/test_persisters/test_ogc_features.py | 58 ++++++++++++- 6 files changed, 242 insertions(+), 24 deletions(-) diff --git a/backend/persisters/ogc_features.py b/backend/persisters/ogc_features.py index fa46c32..0f8344a 100644 --- a/backend/persisters/ogc_features.py +++ b/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 ```` (latest value), + ``_units``, and ``_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, diff --git a/orchestration/assets/products.py b/orchestration/assets/products.py index dc7efd0..d981300 100644 --- a/orchestration/assets/products.py +++ b/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,6 +101,7 @@ 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, @@ -82,17 +109,24 @@ def _build_source_asset(product: dict, source_key: str, group: str): 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 = [] diff --git a/orchestration/config/products.yaml b/orchestration/config/products.yaml index 7e5a2fc..bff1be7 100644 --- a/orchestration/config/products.yaml +++ b/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: [] diff --git a/orchestration/definitions.py b/orchestration/definitions.py index c404ac9..cdc0f36 100644 --- a/orchestration/definitions.py +++ b/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: diff --git a/orchestration/resources/die_config.py b/orchestration/resources/die_config.py index cd572a0..75b914e 100644 --- a/orchestration/resources/die_config.py +++ b/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 diff --git a/tests/test_persisters/test_ogc_features.py b/tests/test_persisters/test_ogc_features.py index 969b8a5..c25bd69 100644 --- a/tests/test_persisters/test_ogc_features.py +++ b/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]