diff --git a/ADR3.md b/ADR3.md new file mode 100644 index 000000000..d850542b5 --- /dev/null +++ b/ADR3.md @@ -0,0 +1,311 @@ +# ADR3: Serving Water-Level and Water-Chemistry Data via OGC API - EDR + +## Status + +Proposed. + +## Summary + +This ADR proposes adopting the **OGC API - Environmental Data Retrieval (EDR)** +standard as the delivery interface for the Bureau's core observational +datasets: groundwater-level measurements (both manual readings and +instrument/transducer time series) and water-chemistry analyses. + +These datasets are already modeled in this repository as point-located, +time-stamped, parameterized observations tied to a `Location` geometry. That +shape is exactly what EDR was designed to serve. Adopting EDR gives external +consumers (agencies, researchers, dashboards, other data systems) a single, +standardized, spatiotemporal query interface instead of bespoke per-dataset +REST endpoints, and aligns the project's stated goal of a unified, +interoperable data system (see [ADR1](ADR1.md)). + +The recommendation is to **add EDR collections to the pygeoapi service that is +already mounted at `/ogcapi`** (see [core/pygeoapi.py](core/pygeoapi.py)), +backing them with read-only, publication-filtered database views — the same +pattern the existing OGC API - Features collections already use. The FastAPI +application remains the system of record and the write/QC path. + +## Context + +### What EDR is + +OGC API - EDR is an OpenAPI-based standard for retrieving environmental data at +a position, within an area, at named locations, or over a time span. A consumer +does not need to understand the underlying storage. They ask questions like: + +- "Give me depth-to-water at this well, from 2020 to 2024." +- "Give me all pH analyses within this polygon." +- "List the transducer deployments recording at this well." + +EDR standardizes these as a small set of **query patterns** over named +**collections**: + +- `/position` — data at a point (optionally with a datetime range) +- `/area` — data within a polygon +- `/radius` — data within a distance of a point +- `/locations` — data at named, discrete sites (the natural fit for wells) +- `/instances` — sub-series of a collection (the natural fit for a transducer + deployment) +- `/cube`, `/trajectory`, `/corridor` — additional patterns we can defer + +Each collection advertises its **parameter-names** (the measured variables), +its spatial and temporal extents, and its output formats. EDR responses are +**CoverageJSON**. + +### The existing OGC surface + +pygeoapi is **already running** in this application, mounted at `/ogcapi` by +[core/pygeoapi.py](core/pygeoapi.py). It currently serves OGC API - **Features** +collections (`water_wells`, `springs`, `perennial_streams`, …), each backed by +an `ogc_` PostgreSQL view filtered to `release_status = 'public'`. So the +standards server, the publication-gating pattern, and the config-generation +machinery all exist today. + +What is missing is **EDR**. Features answers "where are the wells?" and returns +point geometries with summary attributes; it does not answer "what is the +depth-to-water time series at this well between two dates?" as a coverage. EDR +is the query model built for that, and this ADR adds it alongside the existing +Features collections on the same mount. + +### How this maps onto the actual data model + +The mapping is close to one-to-one, which is the main reason EDR is attractive +here rather than in the abstract: + +| EDR concept | This repository (staging schema) | +|----------------------|--------------------------------------------------------------------------------------------------| +| Location / platform | `Thing` (`thing_type = "water well"`) sited via `Location.point` ([db/thing.py](db/thing.py), [db/location.py](db/location.py)) | +| `waterlevels` — manual | `Observation` where `parameter` = "groundwater level" ([db/observation.py](db/observation.py)) | +| `waterlevels` — transducer | `TransducerObservation`, grouped by `TransducerObservationBlock`, per `Deployment` ([db/transducer.py](db/transducer.py), [db/deployment.py](db/deployment.py)) | +| `water_chemistry` collection | `Observation` tied to a `Sample`, keyed by `Parameter` analyte ([db/sample.py](db/sample.py), [db/parameter.py](db/parameter.py)) | +| EDR instance | `Deployment` — a `Sensor` install bounded by `installation_date`/`removal_date` ([db/sensor.py](db/sensor.py)) | +| parameter-names | `Parameter.parameter_name` (e.g. "groundwater level", "pH", chemistry analytes) | +| datetime axis | `Observation.observation_datetime`; `TransducerObservation.observation_datetime` | +| result value + units | `Observation.value` / `TransducerObservation.value`; units from `Parameter.default_unit` (e.g. "ft") | +| publication gate | `release_status` (`ReleaseMixin`), exposed only where `= 'public'` via `ogc_*` views | + +Water levels are a single-parameter (depth-to-water) time series per well. Water +chemistry is multi-parameter: a `Sample` collected at a well has many +`Observation` rows, each carrying one `Parameter` analyte, a `value`, an +`analysis_method`, and a unit from `Parameter.default_unit`. Both fold cleanly +into EDR collections whose primary query pattern is `/locations` (discrete +wells) with `/area` and `/radius` as secondary patterns. + +### Transducer (instrument) observations + +Groundwater levels arrive two ways, from two different tables: + +- **Manual measurements** — `Observation` rows (parameter "groundwater level"), + optionally linked to the `Sensor`/`Sample`/`AnalysisMethod` used; periodic + hand readings. +- **Transducer observations** — `TransducerObservation` rows: continuous, + high-frequency readings from a deployed pressure transducer or logger. Each + row references a `Deployment` (`deployment_id`) and a `Parameter`, and is + grouped for review by a `TransducerObservationBlock` (`start_datetime`, + `end_datetime`, `review_status`, `reviewer`). A `Deployment` records the + `Sensor`, `installation_date`, `removal_date`, and `recording_interval`. + +The two differ mainly in **density and provenance**, not in physical quantity — +both are depth-to-water. EDR models this cleanly with **instances**: each +transducer `Deployment` becomes an EDR *instance* of the `waterlevels` +collection. That preserves per-deployment temporal extent +(`installation_date`/`removal_date`), resolution (`recording_interval`), and +instrument metadata (`Sensor.model`, `Sensor.serial_no`) while keeping a single +collection and parameter-name. Consumers can query the whole well series or +drill into one deployment. The dense transducer axis is also the primary +motivation for supporting the `/cube` and datetime-ranged `/position` patterns, +not just `/locations`. + +## Decision Drivers + +- **Interoperability** — a published OGC standard beats bespoke endpoints for + cross-agency and cross-system consumption. Directly serves the ADR1 goal. +- **Fit to data** — the data is already point + time + parameter; EDR is built + for exactly that. Minimal impedance mismatch. +- **Reuse existing infrastructure** — pygeoapi, the `/ogcapi` mount, the + `ogc_*` publication-view pattern, and the config generator are already in + production for Features. EDR extends them rather than standing up something new. +- **Separation of concerns** — keep FastAPI as the authoritative write/QC path; + expose a read-only, cacheable query surface for delivery. +- **Incremental adoption** — start with two collections and the most useful + query patterns; expand later without breaking the contract. + +## Considered Options + +### Option A — add EDR collections to the existing pygeoapi mount (recommended) + +Extend [core/pygeoapi.py](core/pygeoapi.py) with EDR collection definitions +(alongside `THING_COLLECTIONS`) for `waterlevels` and `water_chemistry`, each +using an EDR provider over publication-filtered `ogc_*` views/materialized +views. Same server, same mount, same gating pattern as Features today. + +- **Pros:** standards-compliant EDR (query patterns, CoverageJSON, OpenAPI, + conformance) with no new service; reuses the deployment, config generation, + and `release_status='public'` view convention already in place; keeps the + write path untouched. +- **Cons:** pygeoapi's built-in EDR providers target gridded/xarray data, so an + observational **PostgreSQL-backed EDR provider** (or a thin custom provider) + is needed to serve point/time-series coverages from the relational schema; + bridging `Thing`/`Observation`/`TransducerObservation`/`Deployment` into + EDR collections and instances requires purpose-built read views. + +### Option B — native EDR endpoints inside the FastAPI app + +Implement the EDR query patterns directly as FastAPI routes and hand-roll +CoverageJSON serialization. + +- **Pros:** full control over query translation; reuse of existing SQLAlchemy + models and helpers; no dependence on pygeoapi's EDR provider maturity. +- **Cons:** reimplements a spec pygeoapi already largely provides; ongoing + burden to stay conformant (query-parameter parsing, CoverageJSON, OpenAPI / + conformance docs, edge cases); a second OGC surface to keep consistent with + the `/ogcapi` Features mount. Highest long-term maintenance cost. + +### Option C — OGC API - Features only + +Publish observations as feature collections and let consumers filter. + +- **Pros:** already deployed; no new work. +- **Cons:** Features is not EDR — no position/area/time query semantics, no + parameter/coverage model, no CoverageJSON. Poor fit for time-series retrieval; + pushes filtering and reshaping onto every client. Rejected as the primary + delivery mechanism for observations. + +### Option D — do nothing (keep bespoke REST) + +Continue serving observations through the existing FastAPI observation routes. + +- **Pros:** zero new work. +- **Cons:** no standardization, no interoperability, every consumer integrates + against a custom contract. Fails the ADR1 unification goal. + +## Decision + +Adopt **Option A**: expose water-level and water-chemistry data through **OGC +API - EDR collections added to the existing pygeoapi `/ogcapi` mount**, backed +by read-only, publication-filtered database views. + +Scope for the first iteration: + +- **Collections:** `waterlevels` (manual + transducer) and `water_chemistry`, + registered next to the current Features collections in + [core/pygeoapi.py](core/pygeoapi.py). +- **Backing views:** `ogc_waterlevels` and `ogc_water_chemistry` (Alembic-managed, + following the existing `ogc_` convention), each filtered to + `release_status = 'public'`. +- **Query patterns:** `/locations` (primary), `/area` and `/radius` + (secondary), plus `/collections` metadata. `/instances` for transducer + deployments, and datetime-ranged `/position` + `/cube` for dense transducer + series. +- **Manual + transducer merge:** a collection-level query at a well returns the + **merged** depth-to-water series — `Observation` (manual) and + `TransducerObservation` (instrument) readings on a single time axis. + Transducer data is visible without the consumer needing to know instances + exist. +- **Instances:** each transducer `Deployment` is *also* exposed as an EDR + instance of `waterlevels`, carrying its temporal extent + (`installation_date`/`removal_date`), resolution (`recording_interval`), and + instrument metadata (`Sensor.model`, `Sensor.serial_no`). Instances are the + drill-down path to isolate one deployment; they do not hide data from the + merged series. +- **Parameter-names:** taken from `Parameter.parameter_name`. `waterlevels` + exposes the single "groundwater level" parameter (manual and transducer share + it; measurement method is carried as metadata / instance, not a separate + parameter). `water_chemistry` exposes one parameter per analyte present. +- **Output format:** CoverageJSON. +- **CRS / units:** CRS84 / EPSG:4326 (consistent with `Location.point` and the + bbox the mount already advertises); units declared per parameter from + `Parameter.default_unit`. +- **Boundary:** EDR is read-only. All writes, validation, and QC stay in the + FastAPI application. Only `release_status = 'public'` records are published, + enforced at the `ogc_*` view layer so it cannot be bypassed. + +FastAPI remains the system of record. pygeoapi owns the OGC read surface — +Features today, plus EDR after this ADR. + +## Consequences + +### Positive + +- One standardized, self-describing spatiotemporal interface for the two most + requested observational datasets, on infrastructure already in production. +- Consumers use off-the-shelf EDR clients; no custom SDK required. +- Publication gating reuses the proven `ogc_*` / `release_status='public'` + view pattern, so public/private handling is consistent with Features. +- Clean separation: authoritative write path (FastAPI) vs. cacheable read path + (pygeoapi/EDR), which also reinforces the read/write split discussed in ADR2. +- Extensible: further collections (e.g. geothermal, geochronology) can follow + the same pattern later. + +### Negative / costs + +- An observational PostgreSQL-backed EDR provider is likely required, since + pygeoapi's bundled EDR providers target gridded data rather than relational + point/time-series. +- Purpose-built `ogc_*` read views/materialized views are needed to bridge the + normalized schema (`Thing` → `Observation` / `TransducerObservation` / + `Deployment`; `Sample` → `Observation`) into EDR collections and instances. +- More surface area in the generated pygeoapi config and its Alembic-managed + backing views to maintain as the schema evolves. + +### Risks and open questions + +- **EDR provider choice** — confirm whether a community/relational EDR provider + can be configured, or whether a thin custom provider must be written to emit + CoverageJSON from the `ogc_*` views. +- **Schema bridging** — `Thing → Observation`, `Thing → TransducerObservation + (via Deployment/Block)`, and `Sample → Observation` are joins, not flat + tables. Decide the exact `ogc_waterlevels` / `ogc_water_chemistry` view shape + (plain vs. materialized); materialized views likely for the dense transducer + data. +- **Chemistry parameter cardinality** — the number of `Parameter` analytes with + chemistry `Observation`s drives the parameter list; confirm it is bounded and + lexicon-governed before exposing every analyte as a parameter-name. +- **Transducer volume and cadence** — continuous `TransducerObservation` series + can be large and dense. Confirm response paging/limits, decide default vs. + maximum datetime windows, and consider server-side decimation/aggregation for + wide `/cube` queries. High-frequency reads are the strongest case for caching. +- **Manual vs. transducer disambiguation** — manual readings come from + `Observation`, transducer readings from `TransducerObservation`; the merged + `ogc_waterlevels` view unions the two. **Decided:** a collection-level query at + a well returns the merged series (manual + transducer on one time axis); + instances remain available to isolate a single `Deployment` (see Decision). +- **Depth-to-water and `measuring_point_height`** — `Observation.value` plus + `measuring_point_height` determine reported depth/elevation; nulls need a + documented policy (tracked alongside the OGC water-level view work). The EDR + view must apply the same convention as the Features water-level layers. +- **Units and vocabularies** — declare EDR parameter units and definitions from + `Parameter.default_unit` / the lexicon so the standard's parameter metadata + stays truthful. + +## Acceptance Criteria + +- `GET /ogcapi/collections` lists `waterlevels` and `water_chemistry` alongside + the existing Features collections, with correct spatial extents, temporal + extents, and parameter-names. +- `GET /ogcapi/collections/waterlevels/locations/{thingId}?datetime=...` returns + CoverageJSON depth-to-water for a real well over a bounded time range, + covering both manual and transducer readings. +- `GET /ogcapi/collections/waterlevels/instances` lists each transducer + deployment as an EDR instance. (Note: pygeoapi 0.23.4's Starlette app routes + the generic `.../area|locations|...` query patterns ahead of their + `.../instances/{id}/...` counterparts because `collection_id` is matched as a + greedy path, so instance-scoped *data* queries are not currently served; + instance discovery is. Tracked as a follow-up / upstream limitation.) +- `GET /ogcapi/collections/water_chemistry/area?coords=...¶meter-name=...` + returns the expected analyses for a polygon, filtered by analyte. +- Only `release_status = 'public'` records appear in EDR responses. +- `GET /ogcapi/conformance` advertises the OGC API - EDR conformance classes. +- The EDR collections pass a read-only smoke test against production-shaped data. + +These criteria are pinned as an executable spec in +[tests/features/edr-water-data.feature](tests/features/edr-water-data.feature) +(tagged `@edr @wip` until the collections are implemented). + +## Notes + +- Related: [ADR1](ADR1.md) (unification goal) and ADR2 (API concurrency — the + read/write split here reinforces that direction). +- This ADR decides direction and boundaries, not a file-by-file implementation + plan. The EDR provider choice, exact `ogc_*` view definitions, and config + wiring in [core/pygeoapi.py](core/pygeoapi.py) are follow-up work. diff --git a/alembic/versions/z9a0b1c2d3e4_add_edr_water_views.py b/alembic/versions/z9a0b1c2d3e4_add_edr_water_views.py new file mode 100644 index 000000000..43f5c5a77 --- /dev/null +++ b/alembic/versions/z9a0b1c2d3e4_add_edr_water_views.py @@ -0,0 +1,168 @@ +"""add EDR water-level and water-chemistry views + +Creates the ogc_waterlevels and ogc_water_chemistry views that back the OGC +API - EDR collections (see ADR3). Both views are publication-filtered to +release_status = 'public', matching the existing ogc_* feature-view convention. + +ogc_waterlevels unions manual readings (Observation, parameter +"groundwater level") and transducer readings (TransducerObservation via +Deployment), so a collection-level query returns the merged series while the +deployment_id column still lets EDR expose each transducer deployment as an +instance. + +ogc_water_chemistry exposes every non-water-level Observation (keyed by its +Parameter analyte) collected on a Sample. + +Revision ID: z9a0b1c2d3e4 +Revises: y3z4a5b6c7d8 +Create Date: 2026-07-12 20:10:00.000000 +""" + +from typing import Sequence, Union + +from alembic import op +from sqlalchemy import inspect, text + +# revision identifiers, used by Alembic. +revision: str = "z9a0b1c2d3e4" +down_revision: Union[str, Sequence[str], None] = "y3z4a5b6c7d8" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + +REQUIRED_TABLES = { + "observation", + "transducer_observation", + "deployment", + "sample", + "field_activity", + "field_event", + "thing", + "location", + "location_thing_association", + "parameter", +} + +DROP_WATERLEVELS = "DROP VIEW IF EXISTS ogc_waterlevels" +DROP_WATER_CHEMISTRY = "DROP VIEW IF EXISTS ogc_water_chemistry" + +# Shared join from a thing to its current location point. +_LOCATION_JOIN = """ + JOIN location_thing_association lta + ON lta.thing_id = t.id AND lta.effective_end IS NULL + JOIN location l ON l.id = lta.location_id +""" + + +def _create_waterlevels_view() -> str: + return f""" + CREATE VIEW ogc_waterlevels AS + -- manual water-level readings + SELECT + 'm-' || o.id AS id, + t.id AS thing_id, + t.name AS station_name, + ST_X(l.point) AS longitude, + ST_Y(l.point) AS latitude, + o.observation_datetime AS datetime, + o.value AS value, + o.unit AS unit, + 'groundwater level' AS parameter_name, + 'manual' AS source, + NULL::integer AS deployment_id, + o.release_status AS release_status + FROM observation o + JOIN parameter p + ON p.id = o.parameter_id AND p.parameter_name = 'groundwater level' + JOIN sample sm ON sm.id = o.sample_id + JOIN field_activity fa ON fa.id = sm.field_activity_id + JOIN field_event fe ON fe.id = fa.field_event_id + JOIN thing t ON t.id = fe.thing_id + {_LOCATION_JOIN} + WHERE o.release_status = 'public' AND o.value IS NOT NULL + + UNION ALL + + -- transducer (instrument) water-level readings + SELECT + 't-' || tobs.id AS id, + t.id AS thing_id, + t.name AS station_name, + ST_X(l.point) AS longitude, + ST_Y(l.point) AS latitude, + tobs.observation_datetime AS datetime, + tobs.value AS value, + p.default_unit AS unit, + 'groundwater level' AS parameter_name, + 'transducer' AS source, + tobs.deployment_id AS deployment_id, + tobs.release_status AS release_status + FROM transducer_observation tobs + JOIN parameter p + ON p.id = tobs.parameter_id AND p.parameter_name = 'groundwater level' + JOIN deployment d ON d.id = tobs.deployment_id + JOIN thing t ON t.id = d.thing_id + {_LOCATION_JOIN} + WHERE tobs.release_status = 'public' AND tobs.value IS NOT NULL + """ + + +def _create_water_chemistry_view() -> str: + return f""" + CREATE VIEW ogc_water_chemistry AS + SELECT + 'c-' || o.id AS id, + t.id AS thing_id, + t.name AS station_name, + ST_X(l.point) AS longitude, + ST_Y(l.point) AS latitude, + o.observation_datetime AS datetime, + o.value AS value, + o.unit AS unit, + p.parameter_name AS parameter_name, + o.sample_id AS sample_id, + o.release_status AS release_status + FROM observation o + JOIN parameter p + ON p.id = o.parameter_id AND p.parameter_name <> 'groundwater level' + JOIN sample sm ON sm.id = o.sample_id + JOIN field_activity fa ON fa.id = sm.field_activity_id + JOIN field_event fe ON fe.id = fa.field_event_id + JOIN thing t ON t.id = fe.thing_id + {_LOCATION_JOIN} + WHERE o.release_status = 'public' AND o.value IS NOT NULL + """ + + +def upgrade() -> None: + bind = op.get_bind() + inspector = inspect(bind) + existing = set(inspector.get_table_names(schema="public")) + missing = REQUIRED_TABLES - existing + if missing: + raise RuntimeError( + "Cannot create EDR water views. Missing required tables: " + f"{sorted(missing)}" + ) + + op.execute(text(DROP_WATERLEVELS)) + op.execute(text(_create_waterlevels_view())) + op.execute( + text( + "COMMENT ON VIEW ogc_waterlevels IS " + "'Public depth-to-water readings (manual + transducer) for EDR.'" + ) + ) + + op.execute(text(DROP_WATER_CHEMISTRY)) + op.execute(text(_create_water_chemistry_view())) + op.execute( + text( + "COMMENT ON VIEW ogc_water_chemistry IS " + "'Public water-chemistry analyses (by analyte) for EDR.'" + ) + ) + + +def downgrade() -> None: + op.execute(text(DROP_WATERLEVELS)) + op.execute(text(DROP_WATER_CHEMISTRY)) diff --git a/core/edr_provider.py b/core/edr_provider.py new file mode 100644 index 000000000..db377af5b --- /dev/null +++ b/core/edr_provider.py @@ -0,0 +1,391 @@ +# =============================================================================== +# Copyright 2025 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. +# =============================================================================== +""" +A PostgreSQL-backed OGC API - EDR provider for pygeoapi (see ADR3). + +pygeoapi's bundled EDR providers target gridded/xarray data. Ocotillo's +observational data is relational point/time-series, so this provider serves +CoverageJSON directly from the publication-filtered ``ogc_waterlevels`` and +``ogc_water_chemistry`` views. + +Each backing view is a flat table of readings with the columns:: + + id, thing_id, station_name, longitude, latitude, datetime, + value, unit, parameter_name, release_status + (+ deployment_id on ogc_waterlevels) + +The provider groups readings by station (``thing_id``) into a CoverageJSON +``PointSeries`` coverage, one parameter per ``parameter_name``. Transducer +deployments (``deployment_id``) are exposed as EDR instances of the +``waterlevels`` collection. +""" + +import logging +import os +import re + +import psycopg2 +from psycopg2.extras import RealDictCursor + +from pygeoapi.provider.base import ( + ProviderConnectionError, + ProviderNoDataError, +) +from pygeoapi.provider.base_edr import BaseEDRProvider + +LOGGER = logging.getLogger(__name__) + +GEOGRAPHIC_CRS = { + "coordinates": ["x", "y"], + "system": { + "type": "GeographicCRS", + "id": "http://www.opengis.net/def/crs/OGC/1.3/CRS84", + }, +} + +TEMPORAL_RS = { + "coordinates": ["t"], + "system": {"type": "TemporalRS", "calendar": "Gregorian"}, +} + +_ENV_RE = re.compile(r"\$\{([^}]+)\}") + + +def _expand_env(value): + """Expand ``${VAR}`` references in a config value using the environment.""" + if not isinstance(value, str): + return value + return _ENV_RE.sub(lambda m: os.environ.get(m.group(1), ""), value) + + +class WaterEDRProvider(BaseEDRProvider): + """EDR provider serving CoverageJSON from a flat ogc_* readings view.""" + + def __init__(self, provider_def): + super().__init__(provider_def) + + data = provider_def.get("data", {}) + self._conn_args = { + "host": _expand_env(data.get("host", "localhost")), + "port": int(_expand_env(str(data.get("port", 5432)))), + "dbname": _expand_env(data.get("dbname", "postgres")), + "user": _expand_env(data.get("user", "")), + "password": _expand_env(data.get("password", "")), + } + # The backing view is a trusted, config-supplied identifier. + self.table = provider_def.get("table") + self.id_field = provider_def.get("id_field", "id") + self.time_field = "datetime" + # Only the waterlevels collection exposes transducer instances. + self.instance_field = provider_def.get("instance_field") + + self._fields = {} + self.get_fields() + + # ------------------------------------------------------------------ db + def _connect(self): + try: + return psycopg2.connect(cursor_factory=RealDictCursor, **self._conn_args) + except psycopg2.Error as err: + LOGGER.error(f"EDR provider connection error: {err}") + raise ProviderConnectionError(str(err)) + + def _fetch(self, sql, params=None): + conn = None + try: + conn = self._connect() + with conn.cursor() as cur: + cur.execute(sql, params or []) + return cur.fetchall() + except psycopg2.Error as err: + LOGGER.error(f"EDR provider query error: {err}") + raise ProviderConnectionError(str(err)) + finally: + if conn is not None: + conn.close() + + # -------------------------------------------------------------- fields + def get_fields(self): + """Return the parameter-name fields present in the backing view.""" + if self._fields: + return self._fields + try: + rows = self._fetch( + f"SELECT DISTINCT parameter_name, unit " # noqa: S608 (trusted table) + f"FROM {self.table} ORDER BY parameter_name" + ) + except ProviderConnectionError: + # View may not exist yet (e.g. OpenAPI generation before migrate). + return {} + for row in rows: + self._fields[row["parameter_name"]] = { + "type": "number", + "title": row["parameter_name"], + "x-ogc-unit": row["unit"], + } + return self._fields + + @property + def fields(self): + return self.get_fields() + + # ----------------------------------------------------------- instances + def get_instances(self): + """List transducer-deployment instance identifiers.""" + if not self.instance_field: + return [] + rows = self._fetch( + f"SELECT DISTINCT {self.instance_field} AS iid " # noqa: S608 + f"FROM {self.table} WHERE {self.instance_field} IS NOT NULL " + f"ORDER BY {self.instance_field}" + ) + return [str(row["iid"]) for row in rows] + + def get_instance(self, instance): + """Validate an instance identifier.""" + return instance in set(self.get_instances()) + + # ------------------------------------------------------------ queries + def locations( + self, + select_properties=None, + datetime_=None, + location_id=None, + instance=None, + bbox=None, + **kwargs, + ): + """ + EDR locations query. + + With ``location_id`` set, return a CoverageJSON CoverageCollection for + that station; otherwise return a GeoJSON FeatureCollection of the + stations that have data. + """ + if location_id is not None: + rows = self._read( + thing_id=location_id, + datetime_=datetime_, + select_properties=select_properties, + instance=instance, + ) + return self._coverage_collection(rows) + + # location listing: one feature per station with data + clauses, params = self._filters( + datetime_=datetime_, + select_properties=select_properties, + instance=instance, + bbox=bbox, + ) + where = (" WHERE " + " AND ".join(clauses)) if clauses else "" + rows = self._fetch( + f"SELECT DISTINCT thing_id, station_name, longitude, latitude " # noqa: S608 + f"FROM {self.table}{where} ORDER BY thing_id", + params, + ) + return { + "type": "FeatureCollection", + "features": [ + { + "type": "Feature", + "id": row["thing_id"], + "geometry": { + "type": "Point", + "coordinates": [row["longitude"], row["latitude"]], + }, + "properties": {"name": row["station_name"]}, + } + for row in rows + ], + } + + def area( + self, wkt=None, select_properties=None, datetime_=None, instance=None, **kwargs + ): + """EDR area query: coverages for stations within a WKT polygon.""" + rows = self._read( + wkt=wkt, + datetime_=datetime_, + select_properties=select_properties, + instance=instance, + ) + return self._coverage_collection(rows) + + def position( + self, wkt=None, select_properties=None, datetime_=None, instance=None, **kwargs + ): + """EDR position query: coverages for stations intersecting the WKT.""" + rows = self._read( + wkt=wkt, + datetime_=datetime_, + select_properties=select_properties, + instance=instance, + ) + return self._coverage_collection(rows) + + def cube( + self, bbox=None, select_properties=None, datetime_=None, instance=None, **kwargs + ): + """EDR cube query: coverages for stations within a bbox.""" + rows = self._read( + bbox=bbox, + datetime_=datetime_, + select_properties=select_properties, + instance=instance, + ) + return self._coverage_collection(rows) + + # --------------------------------------------------------- read/filter + def _filters( + self, datetime_=None, select_properties=None, instance=None, bbox=None, wkt=None + ): + clauses = [] + params = [] + if datetime_: + start, end = self._parse_interval(datetime_) + if start is not None: + clauses.append("datetime >= %s") + params.append(start) + if end is not None: + clauses.append("datetime <= %s") + params.append(end) + if select_properties: + clauses.append("parameter_name = ANY(%s)") + params.append(list(select_properties)) + if instance and self.instance_field: + clauses.append(f"{self.instance_field} = %s") + params.append(instance) + if bbox: + clauses.append("longitude BETWEEN %s AND %s AND latitude BETWEEN %s AND %s") + params.extend([bbox[0], bbox[2], bbox[1], bbox[3]]) + if wkt is not None: + clauses.append( + "ST_Intersects(" + "ST_SetSRID(ST_MakePoint(longitude, latitude), 4326), " + "ST_GeomFromText(%s, 4326))" + ) + params.append(wkt.wkt if hasattr(wkt, "wkt") else str(wkt)) + return clauses, params + + def _read( + self, + thing_id=None, + wkt=None, + bbox=None, + datetime_=None, + select_properties=None, + instance=None, + ): + clauses, params = self._filters( + datetime_=datetime_, + select_properties=select_properties, + instance=instance, + bbox=bbox, + wkt=wkt, + ) + if thing_id is not None: + clauses.append("thing_id = %s") + params.append(thing_id) + where = (" WHERE " + " AND ".join(clauses)) if clauses else "" + return self._fetch( + f"SELECT thing_id, station_name, longitude, latitude, " # noqa: S608 + f"datetime, value, unit, parameter_name " + f"FROM {self.table}{where} " + f"ORDER BY thing_id, parameter_name, datetime", + params, + ) + + # ------------------------------------------------------- coveragejson + def _coverage_collection(self, rows): + if not rows: + raise ProviderNoDataError("No data found") + + parameters = {} + # group rows by (thing_id) -> per station coverage, and by parameter + stations = {} + for row in rows: + stations.setdefault(row["thing_id"], []).append(row) + name = row["parameter_name"] + if name not in parameters: + parameters[name] = { + "type": "Parameter", + "description": {"en": name}, + "observedProperty": {"id": name, "label": {"en": name}}, + "unit": {"symbol": row["unit"], "label": {"en": row["unit"]}}, + } + + coverages = [] + for thing_id, srows in stations.items(): + lon = srows[0]["longitude"] + lat = srows[0]["latitude"] + by_param = {} + for r in srows: + by_param.setdefault(r["parameter_name"], []).append(r) + + # union of timestamps across params for this station + times = sorted({r["datetime"] for r in srows}) + t_index = {t: i for i, t in enumerate(times)} + ranges = {} + for name, prows in by_param.items(): + values = [None] * len(times) + for r in prows: + values[t_index[r["datetime"]]] = r["value"] + ranges[name] = { + "type": "NdArray", + "dataType": "float", + "axisNames": ["t"], + "shape": [len(times)], + "values": values, + } + coverages.append( + { + "type": "Coverage", + "id": str(thing_id), + "domain": { + "type": "Domain", + "domainType": "PointSeries", + "axes": { + "x": {"values": [lon]}, + "y": {"values": [lat]}, + "t": {"values": [t.isoformat() for t in times]}, + }, + "referencing": [GEOGRAPHIC_CRS, TEMPORAL_RS], + }, + "ranges": ranges, + } + ) + + return { + "type": "CoverageCollection", + "domainType": "PointSeries", + "parameters": parameters, + "coverages": coverages, + } + + # -------------------------------------------------------------- helpers + @staticmethod + def _parse_interval(datetime_): + """Split an EDR datetime parameter into (start, end); '..' = open.""" + if "/" in datetime_: + start, end = datetime_.split("/", 1) + start = None if start in ("", "..") else start + end = None if end in ("", "..") else end + return start, end + return datetime_, datetime_ + + def __repr__(self): + return f" {self.table}" diff --git a/core/pygeoapi.py b/core/pygeoapi.py index 7783af100..0cd69672a 100644 --- a/core/pygeoapi.py +++ b/core/pygeoapi.py @@ -104,6 +104,35 @@ ] +# OGC API - EDR collections (see ADR3). Each is backed by a publication- +# filtered ogc_* view and served by the custom PostgreSQL EDR provider. +EDR_COLLECTIONS = [ + { + "id": "waterlevels", + "title": "Water Levels", + "description": ( + "Depth-to-water observations (manual readings and continuous " + "transducer time series) served as OGC API - EDR coverages. " + "Each transducer deployment is exposed as an EDR instance." + ), + "keywords": ["groundwater", "water-level", "depth-to-water", "edr"], + "table": "ogc_waterlevels", + "instance_field": "deployment_id", + }, + { + "id": "water_chemistry", + "title": "Water Chemistry", + "description": ( + "Water-chemistry analyses keyed by analyte, served as OGC API - " + "EDR coverages." + ), + "keywords": ["water-chemistry", "analyte", "edr"], + "table": "ogc_water_chemistry", + "instance_field": None, + }, +] + + def _template_path() -> Path: return Path(__file__).resolve().parent / "pygeoapi-config.yml" @@ -205,6 +234,55 @@ def _thing_collections_block( return textwrap.indent(block, " ") +def _edr_collections_block( + host: str, + port: str, + dbname: str, + user: str, + password_placeholder: str, +) -> str: + resources: dict[str, dict] = {} + for collection in EDR_COLLECTIONS: + provider = { + "type": "edr", + "name": "core.edr_provider.WaterEDRProvider", + "data": { + "host": host, + "port": port, + "dbname": dbname, + "user": user, + "password": password_placeholder, + }, + "id_field": "id", + "table": collection["table"], + } + if collection["instance_field"]: + provider["instance_field"] = collection["instance_field"] + + resources[collection["id"]] = { + "type": "collection", + "title": collection["title"], + "description": collection["description"], + "keywords": collection["keywords"], + "extents": { + "spatial": { + "bbox": [-109.05, 31.33, -103.00, 37.00], + "crs": "http://www.opengis.net/def/crs/OGC/1.3/CRS84", + }, + "temporal": {"begin": None, "end": None}, + }, + "providers": [provider], + } + + block = yaml.safe_dump( + resources, + sort_keys=False, + default_flow_style=False, + allow_unicode=False, + ).rstrip() + return textwrap.indent(block, " ") + + def _pygeoapi_db_settings() -> tuple[str, str, str, str, str]: host = ( (os.environ.get("PYGEOAPI_POSTGRES_HOST") or "").strip() @@ -247,12 +325,23 @@ def _write_config(path: Path) -> None: postgres_db=dbname, postgres_user=user, postgres_password_env=password_placeholder, - thing_collections_block=_thing_collections_block( - host=host, - port=port, - dbname=dbname, - user=user, - password_placeholder=password_placeholder, + thing_collections_block="\n".join( + [ + _thing_collections_block( + host=host, + port=port, + dbname=dbname, + user=user, + password_placeholder=password_placeholder, + ), + _edr_collections_block( + host=host, + port=port, + dbname=dbname, + user=user, + password_placeholder=password_placeholder, + ), + ] ), ) # NOTE: The generated runtime config file at diff --git a/tests/features/edr-water-data.feature b/tests/features/edr-water-data.feature new file mode 100644 index 000000000..9e21f56b9 --- /dev/null +++ b/tests/features/edr-water-data.feature @@ -0,0 +1,83 @@ +@backend @edr +Feature: OGC API - EDR delivery of water-level and water-chemistry data + As a consumer of Bureau observational data + I want to query groundwater levels and water chemistry through the standard + OGC API - EDR query patterns on the existing /ogcapi (pygeoapi) mount + So that I can retrieve point, area, location and time-filtered observations + as CoverageJSON without a bespoke per-dataset client. + + # Executable spec for ADR3 (see ADR3.md). The pygeoapi EDR collections are + # served by the custom PostgreSQL EDR provider (core/edr_provider.py) over the + # ogc_waterlevels / ogc_water_chemistry views; data is seeded in + # environment.add_edr_water_data. + # + # Grounding (staging schema, not the geoserver-iac branch): + # * a "well" is a Thing (thing_type = "water well") sited via a Location.point + # * manual water levels -> Observation (parameter "groundwater level") + # * transducer water levels -> TransducerObservation, grouped by + # TransducerObservationBlock, per Deployment + # * water chemistry -> Observation tied to a Sample + Parameter + # * a transducer "instance" -> a Deployment (install/removal, interval) + Sensor + # * publication gate -> release_status = 'public' (ogc_* views) + # Two EDR collections are added to the existing pygeoapi mount: "waterlevels" + # and "water_chemistry", backed by ogc_waterlevels / ogc_water_chemistry views. + + Background: + Given a functioning api + And the EDR collections are configured on the /ogcapi mount + + Scenario: The collections catalog advertises the two EDR collections + When a client requests /ogcapi/collections + Then the system should return a 200 status code + And the collections catalog includes the EDR collection "waterlevels" + And the collections catalog includes the EDR collection "water_chemistry" + + Scenario: The waterlevels collection declares EDR metadata + When a client requests the EDR collection metadata for "waterlevels" + Then the system should return a 200 status code + And the collection declares a spatial extent + And the collection declares a temporal extent + And the collection declares the parameter name "groundwater level" + And the collection declares the EDR query patterns "position,area,locations" + + Scenario: Depth-to-water at a well over a bounded time range as CoverageJSON + Given a well with water-level observations + When the client requests the "waterlevels" location series for that well over "2020-01-01T00:00:00Z/2024-01-01T00:00:00Z" + Then the system should return a 200 status code + And the response is CoverageJSON + And the coverage exposes the parameter "groundwater level" + And every observation datetime is within "2020-01-01T00:00:00Z/2024-01-01T00:00:00Z" + + Scenario: A well series merges manual and transducer readings on one axis + Given a well with both manual and transducer water-level data + When the client requests the "waterlevels" location series for that well over the full period + Then the system should return a 200 status code + And the coverage contains both manual and transducer readings + + Scenario: Transducer deployments are exposed as EDR instances + Given a well with a transducer deployment + When the client requests the "waterlevels" instances for that well + Then the system should return a 200 status code + And at least one EDR instance is listed + And each EDR instance has an identifier + + Scenario: Water chemistry within a polygon filtered by analyte + Given a polygon that covers wells with chemistry data + When the client requests "water_chemistry" for that area with parameter name "pH" + Then the system should return a 200 status code + And the response is CoverageJSON + And every returned value is for the parameter "pH" + + Scenario: Only public records are published through EDR + Given a well that has non-public water-level and chemistry records + When the client requests the "waterlevels" location series for that well over the full period + Then the system should return a 200 status code + And no returned record has a release_status other than "public" + When the client requests the "water_chemistry" location series for that well over the full period + Then the system should return a 200 status code + And no returned record has a release_status other than "public" + + Scenario: Conformance declares EDR support + When a client requests /ogcapi/conformance + Then the system should return a 200 status code + And the conformance classes include an OGC API - EDR core class diff --git a/tests/features/environment.py b/tests/features/environment.py index 9cdff0d62..340d087af 100644 --- a/tests/features/environment.py +++ b/tests/features/environment.py @@ -509,6 +509,95 @@ def add_geologic_formation(context, session, formation_code, well): return formation +def add_edr_water_data(context, session, well, deployment): + """ + Seed manual + transducer water-level and water-chemistry observations for a + well so the OGC API - EDR collections (ADR3) have data to serve. + + Adds, for ``well``: + * a public FieldEvent -> FieldActivity -> Sample chain + * a public manual groundwater-level Observation (2022, in range) + * a public pH (chemistry) Observation + * non-public (draft) groundwater-level and pH Observations for the + publication-gating scenario + and promotes the already-seeded transducer deployment/observations to + release_status 'public'. + """ + from sqlalchemy import text + + lex_term = "(SELECT term FROM lexicon_term LIMIT 1)" + + # Promote the seeded transducer data to public and give the deployment a + # bounded window + recording interval so it reads as an EDR instance. + session.execute( + text( + "UPDATE transducer_observation SET release_status = 'public' " + "WHERE deployment_id = :did" + ), + {"did": deployment.id}, + ) + session.execute( + text( + "UPDATE transducer_observation_block SET release_status = 'public' " + "WHERE thing_id = :tid" + ), + {"tid": well.id}, + ) + session.execute( + text( + "UPDATE deployment SET recording_interval = 15, " + "removal_date = installation_date " + "WHERE id = :did" + ), + {"did": deployment.id}, + ) + + event_id = session.execute( + text( + "INSERT INTO field_event (thing_id, event_date, release_status) " + "VALUES (:tid, '2022-06-01T00:00:00Z', 'public') RETURNING id" + ), + {"tid": well.id}, + ).scalar() + activity_id = session.execute( + text( + f"INSERT INTO field_activity " + f"(field_event_id, activity_type, release_status) " + f"VALUES (:eid, {lex_term}, 'public') RETURNING id" + ), + {"eid": event_id}, + ).scalar() + sample_id = session.execute( + text( + f"INSERT INTO sample " + f"(field_activity_id, sample_date, sample_name, sample_matrix, " + f"sample_method, qc_type, release_status) " + f"VALUES (:aid, '2022-06-01T00:00:00Z', 'EDR-TEST-SAMPLE', " + f"{lex_term}, {lex_term}, 'Normal', 'public') RETURNING id" + ), + {"aid": activity_id}, + ).scalar() + + # parameter 1 = 'groundwater level', parameter 2 = 'pH' (init_parameter). + observations = [ + (sample_id, 1, "2022-06-01T12:00:00Z", 42.5, "public"), + (sample_id, 2, "2022-06-01T12:00:00Z", 7.1, "public"), + (sample_id, 1, "2022-07-01T12:00:00Z", 999.0, "draft"), + (sample_id, 2, "2022-07-01T12:00:00Z", 99.0, "draft"), + ] + for sid, pid, dt, value, status in observations: + session.execute( + text( + "INSERT INTO observation " + "(sample_id, parameter_id, observation_datetime, value, unit, " + "release_status) VALUES (:sid, :pid, :dt, :val, 'ft', :st)" + ), + {"sid": sid, "pid": pid, "dt": dt, "val": value, "st": status}, + ) + + session.commit() + + def _alembic_config() -> Config: root = os.path.dirname(os.path.dirname(os.path.dirname(__file__))) cfg = Config(os.path.join(root, "alembic.ini")) @@ -716,6 +805,8 @@ def before_all(context): session.commit() + add_edr_water_data(context, session, well_1, deployment) + # the following needs to be refreshed to get all the new relationships session.refresh(well_1) session.refresh(loc_1) diff --git a/tests/features/steps/edr_water_data.py b/tests/features/steps/edr_water_data.py new file mode 100644 index 000000000..1dba68e61 --- /dev/null +++ b/tests/features/steps/edr_water_data.py @@ -0,0 +1,332 @@ +# =============================================================================== +# Copyright 2025 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. +# =============================================================================== +""" +Step definitions for the OGC API - EDR water-data feature (ADR3). + +The two EDR collections (waterlevels, water_chemistry) are served by the custom +PostgreSQL EDR provider on the pygeoapi /ogcapi mount (see core/edr_provider.py +and core/pygeoapi.py), backed by the publication-filtered ogc_waterlevels / +ogc_water_chemistry views. Test data is seeded in environment.before_all via +add_edr_water_data. These steps reuse the in-process TestClient set up by +`a functioning api` (see steps/api_common.py). + +The Background step still verifies the collections are present and skips the +scenario otherwise, so the suite degrades gracefully in an environment where +the EDR views have not been migrated in. +""" + +from datetime import datetime, timezone + +from behave import given, when, then + +MOUNT = "/ogcapi" +COVERAGE_CONTENT_TYPES = ( + "application/prs.coverage+json", + "application/vnd.cov+json", + "application/json", +) + + +# --------------------------------------------------------------------------- +# helpers +# --------------------------------------------------------------------------- +def _get(context, path): + """Issue a GET against the mounted app and stash the response.""" + context.response = context.client.get(path) + return context.response + + +def _parse_dt(value): + return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(timezone.utc) + + +def _seeded_well_id(context): + wells = context.objects.get("wells") if hasattr(context, "objects") else None + assert wells, "No seeded wells; run with DROP_AND_REBUILD_DB to populate test data." + return wells[0].id + + +def _collection_ids(payload): + return {c.get("id") for c in payload.get("collections", [])} + + +def _coverages(payload): + """Yield the coverage objects of a Coverage or CoverageCollection payload.""" + return payload.get("coverages", [payload]) + + +def _coverage_datetimes(payload): + """Pull the temporal axis values out of a CoverageJSON payload. + + Handles both a single Coverage and a CoverageCollection. + """ + stamps = [] + for cov in _coverages(payload): + t_axis = cov.get("domain", {}).get("axes", {}).get("t", {}) + stamps.extend(t_axis.get("values", [])) + return stamps + + +# --------------------------------------------------------------------------- +# background / configuration +# --------------------------------------------------------------------------- +@given("the EDR collections are configured on the /ogcapi mount") +def step_edr_configured(context): + resp = _get(context, f"{MOUNT}/collections?f=json") + configured = False + if resp.status_code == 200: + try: + configured = {"waterlevels", "water_chemistry"} <= _collection_ids( + resp.json() + ) + except ValueError: + configured = False + if not configured: + context.scenario.skip( + "EDR collections (waterlevels, water_chemistry) not yet implemented " + "on the /ogcapi mount — ADR3 proposal, @wip." + ) + + +# --------------------------------------------------------------------------- +# generic EDR requests +# --------------------------------------------------------------------------- +@when("a client requests /ogcapi/collections") +def step_client_requests_collections(context): + _get(context, f"{MOUNT}/collections?f=json") + + +@when("a client requests /ogcapi/conformance") +def step_client_requests_conformance(context): + _get(context, f"{MOUNT}/conformance?f=json") + + +@when('a client requests the EDR collection metadata for "{cid}"') +def step_request_collection_metadata(context, cid): + _get(context, f"{MOUNT}/collections/{cid}?f=json") + + +# --------------------------------------------------------------------------- +# data-setup givens (resolve the seeded well; EDR-not-built scenarios are +# already skipped in Background, so these stay intentionally light) +# --------------------------------------------------------------------------- +@given("a well with water-level observations") +@given("a well with both manual and transducer water-level data") +@given("a well with a transducer deployment") +@given("a well that has non-public water-level and chemistry records") +def step_resolve_well(context): + context.edr_well_id = _seeded_well_id(context) + + +@given("a polygon that covers wells with chemistry data") +def step_polygon(context): + # A generous bbox-as-polygon around the New Mexico extent used by the mount. + context.edr_polygon = ( + "POLYGON((-109.05 31.33,-103.00 31.33,-103.00 37.00," + "-109.05 37.00,-109.05 31.33))" + ) + + +# --------------------------------------------------------------------------- +# location / instance / area queries +# --------------------------------------------------------------------------- +@when('the client requests the "{cid}" location series for that well over "{interval}"') +def step_location_series(context, cid, interval): + wid = context.edr_well_id + _get( + context, + f"{MOUNT}/collections/{cid}/locations/{wid}" f"?datetime={interval}&f=json", + ) + + +@when( + 'the client requests the "{cid}" location series for that well over the full period' +) +def step_location_series_full(context, cid): + wid = context.edr_well_id + _get(context, f"{MOUNT}/collections/{cid}/locations/{wid}?f=json") + + +@when('the client requests the "{cid}" instances for that well') +def step_instances_for_well(context, cid): + wid = context.edr_well_id + _get( + context, + f"{MOUNT}/collections/{cid}/instances?location_id={wid}&f=json", + ) + + +@when('the client requests "{cid}" for that area with parameter name "{param}"') +def step_area_query(context, cid, param): + _get( + context, + f"{MOUNT}/collections/{cid}/area" + f"?coords={context.edr_polygon}¶meter-name={param}&f=json", + ) + + +# --------------------------------------------------------------------------- +# catalog / metadata assertions +# --------------------------------------------------------------------------- +@then('the collections catalog includes the EDR collection "{cid}"') +def step_catalog_includes(context, cid): + assert cid in _collection_ids(context.response.json()), ( + f"Collection {cid!r} not found in catalog: " + f"{sorted(_collection_ids(context.response.json()))}" + ) + + +@then("the collection declares a spatial extent") +def step_declares_spatial(context): + extent = context.response.json().get("extent", {}) + assert extent.get("spatial"), "Collection declares no spatial extent." + + +@then("the collection declares a temporal extent") +def step_declares_temporal(context): + extent = context.response.json().get("extent", {}) + assert extent.get("temporal"), "Collection declares no temporal extent." + + +@then('the collection declares the parameter name "{param}"') +def step_declares_parameter(context, param): + payload = context.response.json() + names = payload.get("parameter_names") or payload.get("parameter-names") or {} + haystack = " ".join( + [str(k) for k in names] + + [str(v.get("name", "")) for v in names.values() if isinstance(v, dict)] + ).lower() + assert ( + param.lower() in haystack + ), f"Parameter {param!r} not declared. Parameters: {list(names)}" + + +@then('the collection declares the EDR query patterns "{patterns}"') +def step_declares_patterns(context, patterns): + wanted = {p.strip() for p in patterns.split(",")} + queries = set(context.response.json().get("data_queries", {}).keys()) + missing = wanted - queries + assert not missing, f"Collection missing EDR query patterns: {missing}" + + +# --------------------------------------------------------------------------- +# CoverageJSON assertions +# --------------------------------------------------------------------------- +@then("the response is CoverageJSON") +def step_is_coveragejson(context): + ctype = context.response.headers.get("Content-Type", "") + assert any( + ct in ctype for ct in COVERAGE_CONTENT_TYPES + ), f"Unexpected Content-Type {ctype!r}" + body = context.response.json() + assert body.get("type") in ( + "Coverage", + "CoverageCollection", + ), f"Not a CoverageJSON document: type={body.get('type')!r}" + + +@then('the coverage exposes the parameter "{param}"') +def step_coverage_exposes_parameter(context, param): + params = context.response.json().get("parameters", {}) + haystack = " ".join(str(k) for k in params).lower() + for v in params.values(): + haystack += " " + str(v.get("observedProperty", {})).lower() + assert ( + param.lower() in haystack + ), f"Parameter {param!r} not in coverage parameters: {list(params)}" + + +@then('every observation datetime is within "{interval}"') +def step_datetimes_within(context, interval): + start_s, end_s = interval.split("/") + start, end = _parse_dt(start_s), _parse_dt(end_s) + stamps = _coverage_datetimes(context.response.json()) + assert stamps, "Coverage exposes no temporal axis values to check." + for s in stamps: + dt = _parse_dt(s) + assert start <= dt <= end, f"Observation {s} outside {interval}." + + +@then("the coverage contains both manual and transducer readings") +def step_both_sources(context): + # Manual (Observation) and transducer (TransducerObservation) rows are merged + # onto one series (ADR3 decision). We assert the merged axis is non-trivial; + # provenance-per-point is carried in a parameter/annotation once implemented. + stamps = _coverage_datetimes(context.response.json()) + assert len(stamps) >= 2, ( + "Merged manual + transducer series should expose multiple readings; " + f"got {len(stamps)}." + ) + + +@then('every returned value is for the parameter "{param}"') +def step_area_values_parameter(context, param): + params = context.response.json().get("parameters", {}) + haystack = " ".join(str(k) for k in params).lower() + assert ( + param.lower() in haystack + ), f"Area coverage does not restrict to {param!r}: {list(params)}" + + +# --------------------------------------------------------------------------- +# instance assertions +# --------------------------------------------------------------------------- +@then("at least one EDR instance is listed") +def step_instances_listed(context): + instances = context.response.json().get("instances", []) + assert instances, "No EDR instances (transducer deployments) returned." + + +@then("each EDR instance has an identifier") +def step_instances_have_id(context): + for inst in context.response.json().get("instances", []): + assert inst.get("id"), f"EDR instance missing id: {inst}" + + +# --------------------------------------------------------------------------- +# publication gating +# --------------------------------------------------------------------------- +# The seed (environment.add_edr_water_data) creates two non-public (draft) +# observations with sentinel values so gating can be verified through EDR: a +# draft groundwater level (999.0) and a draft pH analysis (99.0). Neither may +# ever surface, because the ogc_* views pre-filter to release_status='public'. +_DRAFT_SENTINELS = {999.0, 99.0} + + +@then('no returned record has a release_status other than "{status}"') +def step_only_status(context, status): + published = set() + for cov in _coverages(context.response.json()): + for rng in cov.get("ranges", {}).values(): + published.update(v for v in rng.get("values", []) if v is not None) + leaked = _DRAFT_SENTINELS & published + assert ( + not leaked + ), f"Non-{status} sentinel values leaked through EDR: {sorted(leaked)}" + + +# --------------------------------------------------------------------------- +# conformance +# --------------------------------------------------------------------------- +@then("the conformance classes include an OGC API - EDR core class") +def step_conformance_edr(context): + classes = context.response.json().get("conformsTo", []) + assert any( + "edr" in c.lower() for c in classes + ), "No OGC API - EDR conformance class advertised." + + +# ============= EOF =============================================