From 9b28ced81ba444495e9b4bd19df1b0f78baa03e8 Mon Sep 17 00:00:00 2001 From: jakeross Date: Sat, 11 Jul 2026 21:59:05 -0600 Subject: [PATCH 1/5] docs: add ADR3 for serving water data via OGC API - EDR Propose adopting OGC API - EDR as the read-only delivery interface for groundwater-level measurements (manual + transducer time series) and water-chemistry analyses, served by pygeoapi as a facade over the existing PostgreSQL/PostGIS database. Decision: two collections (waterlevels, water-chemistry); transducer deployments (Equipment) exposed as EDR instances; collection-level water-level queries return the merged manual + transducer series. FastAPI stays the system of record; GeoServer keeps Features/WMS. Co-Authored-By: Claude Opus 4.8 --- ADR3.md | 275 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 275 insertions(+) create mode 100644 ADR3.md diff --git a/ADR3.md b/ADR3.md new file mode 100644 index 000000000..280555d41 --- /dev/null +++ b/ADR3.md @@ -0,0 +1,275 @@ +# 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 public delivery interface for the Bureau's core observational +datasets: groundwater-level measurements (both manual readings and +instrument/transducer time series) and water-chemistry analyses. + +Both datasets are already modeled in this repository as point-located, +time-stamped, parameterized observations tied to a `SampleLocation` 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 expose EDR as a **read-only query facade layered over +the existing PostgreSQL/PostGIS database**, keeping the FastAPI application as +the system of record and write path. + +## Context + +### What EDR is + +OGC API - EDR is an OpenAPI-based standard for retrieving environmental data at +a position, within an area, along a trajectory, 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 point, from 2020 to 2024." +- "Give me all nitrate analyses within this polygon." +- "List the locations that have chloride data." + +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) +- `/items` — direct access to individual features +- `/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. Responses are +typically **CoverageJSON** or GeoJSON. + +### How this maps onto the existing 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 | +|----------------------|----------------------------------------------------------------------------------------| +| Location / platform | `Well` → `SampleLocation` (`Geometry(POINT, srid=4326)` in [db/base.py](db/base.py)) | +| `waterlevels` collection | `GroundwaterLevelObservation` via `WellTimeseries` ([db/timeseries.py](db/timeseries.py)) | +| EDR instance | `Equipment` deployment — the transducer/logger recording a series ([db/base.py](db/base.py)) | +| `water-chemistry` collection | `WaterChemistryAnalysis` via `WaterChemistryAnalysisSet` ([db/chemistry.py](db/chemistry.py)) | +| parameter-names | depth-to-water (`value`/`unit`, default `ftbgs`); chemistry `analyte` values | +| datetime axis | `GroundwaterLevelObservation.timestamp`; `WaterChemistryAnalysis.analysis_timestamp` | +| result value + units | `value`, `unit` (unit already normalized against `lexicon_term`) | + +Water levels are a single-parameter (depth-to-water) time series per well. Water +chemistry is a multi-parameter set keyed by `analyte`, where each +`WaterChemistryAnalysisSet` shares a `collection_timestamp` and each child +`WaterChemistryAnalysis` carries its own `analyte`, `value`, `unit`, +`uncertainty`, and `method`. 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, and both live in the same +`GroundwaterLevelObservation` table: + +- **Manual measurements** — periodic hand readings, no instrument attached. +- **Transducer observations** — continuous, high-frequency readings from a + deployed pressure transducer or data logger. These are distinguished by a + non-null `WellTimeseries.equipment_id` pointing at an `Equipment` row whose + `equipment_type` is a transducer/logger, with `recording_interval` (cadence), + `date_installed`, and `date_removed` bounding the deployment. + +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 (`Equipment` bounded by install/removal dates) becomes an +EDR *instance* of the `waterlevels` collection. That preserves per-deployment +temporal extent, resolution (`recording_interval`), and instrument metadata +(`model`, `serial_no`) while keeping a single collection and parameter-name. +Consumers can query the whole well series or drill into one instrument +deployment. The dense transducer axis is also the primary motivation for +supporting the `/cube` and datetime-ranged `/position` patterns, not just +`/locations`. + +### Why now + +This branch introduces GeoServer as spatial infrastructure (see +`geoserver_iac/`). GeoServer covers WMS, WFS, and OGC API - Features well, but +**GeoServer does not implement OGC API - EDR**. Feature access alone does not +give consumers the position/area/time query semantics that observational data +needs. This ADR fills that gap and clarifies the division of labor between +GeoServer (features, maps) and the EDR facade (observations, time series). + +## 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. +- **Separation of concerns** — keep FastAPI as the authoritative write/QC path; + expose a read-only, cacheable query surface for delivery. +- **Standards, not lock-in** — EDR is client-agnostic; any EDR client works. +- **Incremental adoption** — start with two collections and the two most useful + query patterns; expand later without breaking the contract. + +## Considered Options + +### Option A — pygeoapi as an EDR facade over PostgreSQL/PostGIS (recommended) + +Run [pygeoapi](https://pygeoapi.io/) as a separate read-only service configured +with two EDR collections backed by the existing database (via custom EDR +providers, or SQL views shaped for pygeoapi's providers). pygeoapi is a +reference implementation of OGC API - EDR and already appears transitively in +the environment. + +- **Pros:** standards-compliant EDR out of the box (query patterns, + CoverageJSON, OpenAPI, conformance) with little bespoke protocol code; keeps + the write path untouched; deployable alongside GeoServer as another + read service. +- **Cons:** a second service and config surface to operate; custom providers + needed to bridge the normalized schema (well → timeseries → observation) into + EDR's collection/parameter model; two mental models (FastAPI + pygeoapi). + +### Option B — native EDR endpoints inside the existing FastAPI app + +Implement the EDR query patterns directly as FastAPI routes and hand-roll +CoverageJSON serialization. + +- **Pros:** one service, one deployment, one auth story; full control over + query translation and reuse of existing SQLAlchemy models and helpers. +- **Cons:** we reimplement a spec that already has a reference implementation; + ongoing burden to stay conformant (query-parameter parsing, CoverageJSON, + OpenAPI/conformance docs, edge cases). Highest long-term maintenance cost. + +### Option C — GeoServer only (OGC API - Features) + +Publish wells and observations as feature collections and let consumers filter. + +- **Pros:** already on this branch; no new service. +- **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 via `api/timeseries.py` and `api/chemisty.py`. + +- **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 served by pygeoapi as a read-only facade** over the existing +PostgreSQL/PostGIS database. + +Scope for the first iteration: + +- **Collections:** `waterlevels` (manual + transducer) and `water-chemistry`. +- **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 — manual readings and transducer readings on + a single time axis. Transducer data is visible without the consumer needing to + know instances exist. +- **Instances:** each transducer/logger `Equipment` deployment is *also* exposed + as an EDR instance of `waterlevels`, carrying its temporal extent + (`date_installed`/`date_removed`), resolution (`recording_interval`), and + instrument metadata (`model`, `serial_no`). Instances are the drill-down path + to isolate one deployment; they do not hide data from the merged series. +- **Parameter-names:** `waterlevels` exposes a single depth-to-water parameter + (manual and transducer readings share it; measurement method is carried as + metadata / instance, not as a separate parameter); `water-chemistry` exposes + one parameter per `analyte` present in the lexicon. +- **Output formats:** CoverageJSON (primary) and GeoJSON. +- **CRS / units:** EPSG:4326 (consistent with `SampleLocation` and the + project's geopackage SRS convention); units carried from `unit` and declared + per parameter. +- **Boundary:** EDR is read-only. All writes, validation, and QC stay in the + FastAPI application. Only QC-approved / visible records are published + (`WaterChemistryAnalysisSet.visible`, and `quality_control_status` on + observations). + +FastAPI remains the system of record. GeoServer remains responsible for maps +and OGC API - Features. pygeoapi owns the EDR observational surface. + +## Consequences + +### Positive + +- One standardized, self-describing spatiotemporal interface for the two most + requested observational datasets. +- Consumers use off-the-shelf EDR clients; no custom SDK required. +- Clean separation: authoritative write path (FastAPI) vs. cacheable read path + (pygeoapi/EDR), which also helps the concurrency posture discussed in ADR2. +- Extensible: new collections (e.g. geothermal, geochronology) follow the same + pattern later. + +### Negative / costs + +- A new service to deploy, monitor, and secure (align with the existing + GeoServer IaC on this branch). +- Custom EDR providers or purpose-built SQL views are required to bridge the + normalized relational schema into EDR collections and parameters. +- Two frameworks in the delivery stack (FastAPI + pygeoapi) to keep in sync as + the schema evolves. + +### Risks and open questions + +- **Schema bridging** — well → timeseries → observation and set → analysis are + joins, not flat tables. Decide between custom pygeoapi providers vs. dedicated + read views/materialized views. Views are likely simpler to start. +- **Chemistry parameter cardinality** — number of `analyte` values drives the + parameter list; confirm this is bounded and lexicon-governed before exposing + every analyte as a parameter-name. +- **Transducer volume and cadence** — continuous transducer 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 the EDR + facade. +- **Manual vs. transducer disambiguation** — the query that splits the two is + presence of `WellTimeseries.equipment_id` (and transducer `equipment_type`). + Confirm `equipment_type` values are lexicon-governed so instance selection is + reliable. **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 transducer deployment (see Decision). +- **Private-IP database access** — the facade must reach Cloud SQL over the + private IP path (`10.10.0.3`), consistent with the datastore convention noted + for GeoServer; public IP returns 502s. +- **Publication gating** — enforce that only `visible` / QC-approved records + reach EDR, ideally at the view layer so it cannot be bypassed. +- **Units and vocabularies** — declare EDR parameter units and definitions from + the `lexicon_term` table so the standard's parameter metadata stays truthful. + +## Acceptance Criteria + +- `GET /collections` lists `waterlevels` and `water-chemistry` with correct + spatial extents, temporal extents, and parameter-names. +- `GET /collections/waterlevels/locations/{wellId}?datetime=...` returns + CoverageJSON depth-to-water for a real well over a bounded time range, + covering both manual and transducer readings. +- `GET /collections/waterlevels/instances` lists transducer deployments for a + well with correct temporal extents and resolution, and querying one instance + returns only that deployment's dense series. +- `GET /collections/water-chemistry/area?coords=...¶meter-name=...` returns + the expected analyses for a polygon, filtered by analyte. +- Only QC-approved / visible records appear in EDR responses. +- The service reaches Cloud SQL over the private IP and passes a read-only + smoke test against production-shaped data. +- OpenAPI and conformance documents validate against the EDR spec. + +## Notes + +- Related: [ADR1](ADR1.md) (unification goal), ADR2 (API concurrency — the + read/write split here reinforces that direction), and the GeoServer IaC on + the `geoserver-iac` branch (complementary Features/WMS surface). +- This ADR decides direction and boundaries, not a file-by-file implementation + plan. Provider-vs-view choice and deployment wiring are follow-up work. From f3ee4dae23679e781fa01bf4da0e0edb478ac790 Mon Sep 17 00:00:00 2001 From: jakeross Date: Sat, 11 Jul 2026 22:04:08 -0600 Subject: [PATCH 2/5] docs: drop private-IP db access note from ADR3 Remove the Cloud SQL private-IP (10.10.0.3) risk bullet and its matching acceptance-criterion line. Co-Authored-By: Claude Opus 4.8 --- ADR3.md | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/ADR3.md b/ADR3.md index 280555d41..4b460543d 100644 --- a/ADR3.md +++ b/ADR3.md @@ -241,9 +241,6 @@ and OGC API - Features. pygeoapi owns the EDR observational surface. reliable. **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 transducer deployment (see Decision). -- **Private-IP database access** — the facade must reach Cloud SQL over the - private IP path (`10.10.0.3`), consistent with the datastore convention noted - for GeoServer; public IP returns 502s. - **Publication gating** — enforce that only `visible` / QC-approved records reach EDR, ideally at the view layer so it cannot be bypassed. - **Units and vocabularies** — declare EDR parameter units and definitions from @@ -262,8 +259,7 @@ and OGC API - Features. pygeoapi owns the EDR observational surface. - `GET /collections/water-chemistry/area?coords=...¶meter-name=...` returns the expected analyses for a polygon, filtered by analyte. - Only QC-approved / visible records appear in EDR responses. -- The service reaches Cloud SQL over the private IP and passes a read-only - smoke test against production-shaped data. +- The service passes a read-only smoke test against production-shaped data. - OpenAPI and conformance documents validate against the EDR spec. ## Notes From ae62e5cedbc6cf6c89fddba827cface26f34f920 Mon Sep 17 00:00:00 2001 From: jakeross Date: Sun, 12 Jul 2026 08:59:47 -0600 Subject: [PATCH 3/5] docs: correct ADR3 to staging schema and add EDR BDD spec Rewrite ADR3 against the real staging models (Thing/Location, Observation, TransducerObservation/TransducerObservationBlock/Deployment/Sensor, Sample/Parameter) instead of the divergent geoserver-iac branch schema. Key corrections: - EDR is added to the pygeoapi service already mounted at /ogcapi (core/pygeoapi.py), not a new standalone service. - Publication gating uses release_status = 'public' via ogc_* views, matching the existing OGC API - Features pattern. - Parameter names come from Parameter.parameter_name; units from default_unit. - Note that pygeoapi's bundled EDR providers are gridded, so an observational PostgreSQL-backed EDR provider is likely required. Add the executable spec: - tests/features/edr-water-data.feature (@edr @wip, 9 scenarios) - tests/features/steps/edr_water_data.py (reuses the in-process TestClient; Background skips while the EDR collections are unimplemented) Co-Authored-By: Claude Opus 4.8 --- ADR3.md | 343 ++++++++++++----------- tests/features/edr-water-data.feature | 88 ++++++ tests/features/steps/edr_water_data.py | 367 +++++++++++++++++++++++++ 3 files changed, 645 insertions(+), 153 deletions(-) create mode 100644 tests/features/edr-water-data.feature create mode 100644 tests/features/steps/edr_water_data.py diff --git a/ADR3.md b/ADR3.md index 4b460543d..1d96f213a 100644 --- a/ADR3.md +++ b/ADR3.md @@ -7,33 +7,35 @@ Proposed. ## Summary This ADR proposes adopting the **OGC API - Environmental Data Retrieval (EDR)** -standard as the public delivery interface for the Bureau's core observational +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. -Both datasets are already modeled in this repository as point-located, -time-stamped, parameterized observations tied to a `SampleLocation` 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, +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 expose EDR as a **read-only query facade layered over -the existing PostgreSQL/PostGIS database**, keeping the FastAPI application as -the system of record and write path. +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, along a trajectory, or over a time span. A consumer +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 point, from 2020 to 2024." -- "Give me all nitrate analyses within this polygon." -- "List the locations that have chloride data." +- "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**: @@ -42,67 +44,77 @@ EDR standardizes these as a small set of **query patterns** over named - `/area` — data within a polygon - `/radius` — data within a distance of a point - `/locations` — data at named, discrete sites (the natural fit for wells) -- `/items` — direct access to individual features +- `/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. Responses are -typically **CoverageJSON** or GeoJSON. +its spatial and temporal extents, and its output formats. EDR responses are +**CoverageJSON**. -### How this maps onto the existing data model +### 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 | -|----------------------|----------------------------------------------------------------------------------------| -| Location / platform | `Well` → `SampleLocation` (`Geometry(POINT, srid=4326)` in [db/base.py](db/base.py)) | -| `waterlevels` collection | `GroundwaterLevelObservation` via `WellTimeseries` ([db/timeseries.py](db/timeseries.py)) | -| EDR instance | `Equipment` deployment — the transducer/logger recording a series ([db/base.py](db/base.py)) | -| `water-chemistry` collection | `WaterChemistryAnalysis` via `WaterChemistryAnalysisSet` ([db/chemistry.py](db/chemistry.py)) | -| parameter-names | depth-to-water (`value`/`unit`, default `ftbgs`); chemistry `analyte` values | -| datetime axis | `GroundwaterLevelObservation.timestamp`; `WaterChemistryAnalysis.analysis_timestamp` | -| result value + units | `value`, `unit` (unit already normalized against `lexicon_term`) | +| 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 a multi-parameter set keyed by `analyte`, where each -`WaterChemistryAnalysisSet` shares a `collection_timestamp` and each child -`WaterChemistryAnalysis` carries its own `analyte`, `value`, `unit`, -`uncertainty`, and `method`. Both fold cleanly into EDR collections whose -primary query pattern is `/locations` (discrete wells) with `/area` and -`/radius` as secondary patterns. +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, and both live in the same -`GroundwaterLevelObservation` table: +Groundwater levels arrive two ways, from two different tables: -- **Manual measurements** — periodic hand readings, no instrument attached. -- **Transducer observations** — continuous, high-frequency readings from a - deployed pressure transducer or data logger. These are distinguished by a - non-null `WellTimeseries.equipment_id` pointing at an `Equipment` row whose - `equipment_type` is a transducer/logger, with `recording_interval` (cadence), - `date_installed`, and `date_removed` bounding the deployment. +- **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 (`Equipment` bounded by install/removal dates) becomes an -EDR *instance* of the `waterlevels` collection. That preserves per-deployment -temporal extent, resolution (`recording_interval`), and instrument metadata -(`model`, `serial_no`) while keeping a single collection and parameter-name. -Consumers can query the whole well series or drill into one instrument -deployment. The dense transducer axis is also the primary motivation for -supporting the `/cube` and datetime-ranged `/position` patterns, not just -`/locations`. - -### Why now - -This branch introduces GeoServer as spatial infrastructure (see -`geoserver_iac/`). GeoServer covers WMS, WFS, and OGC API - Features well, but -**GeoServer does not implement OGC API - EDR**. Feature access alone does not -give consumers the position/area/time query semantics that observational data -needs. This ADR fills that gap and clarifies the division of labor between -GeoServer (features, maps) and the EDR facade (observations, time series). +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 @@ -110,54 +122,58 @@ GeoServer (features, maps) and the EDR facade (observations, time series). 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. -- **Standards, not lock-in** — EDR is client-agnostic; any EDR client works. -- **Incremental adoption** — start with two collections and the two most useful +- **Incremental adoption** — start with two collections and the most useful query patterns; expand later without breaking the contract. ## Considered Options -### Option A — pygeoapi as an EDR facade over PostgreSQL/PostGIS (recommended) +### Option A — add EDR collections to the existing pygeoapi mount (recommended) -Run [pygeoapi](https://pygeoapi.io/) as a separate read-only service configured -with two EDR collections backed by the existing database (via custom EDR -providers, or SQL views shaped for pygeoapi's providers). pygeoapi is a -reference implementation of OGC API - EDR and already appears transitively in -the environment. +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 out of the box (query patterns, - CoverageJSON, OpenAPI, conformance) with little bespoke protocol code; keeps - the write path untouched; deployable alongside GeoServer as another - read service. -- **Cons:** a second service and config surface to operate; custom providers - needed to bridge the normalized schema (well → timeseries → observation) into - EDR's collection/parameter model; two mental models (FastAPI + pygeoapi). +- **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 existing FastAPI app +### Option B — native EDR endpoints inside the FastAPI app Implement the EDR query patterns directly as FastAPI routes and hand-roll CoverageJSON serialization. -- **Pros:** one service, one deployment, one auth story; full control over - query translation and reuse of existing SQLAlchemy models and helpers. -- **Cons:** we reimplement a spec that already has a reference implementation; - ongoing burden to stay conformant (query-parameter parsing, CoverageJSON, - OpenAPI/conformance docs, edge cases). Highest long-term maintenance cost. +- **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 — GeoServer only (OGC API - Features) +### Option C — OGC API - Features only -Publish wells and observations as feature collections and let consumers filter. +Publish observations as feature collections and let consumers filter. -- **Pros:** already on this branch; no new service. -- **Cons:** Features is not EDR. No position/area/time query semantics, no +- **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 via `api/timeseries.py` and `api/chemisty.py`. +Continue serving observations through the existing FastAPI observation routes. - **Pros:** zero new work. - **Cons:** no standardization, no interoperability, every consumer integrates @@ -166,106 +182,127 @@ Continue serving via `api/timeseries.py` and `api/chemisty.py`. ## Decision Adopt **Option A**: expose water-level and water-chemistry data through **OGC -API - EDR served by pygeoapi as a read-only facade** over the existing -PostgreSQL/PostGIS database. +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`. +- **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 — manual readings and transducer readings on - a single time axis. Transducer data is visible without the consumer needing to - know instances exist. -- **Instances:** each transducer/logger `Equipment` deployment is *also* exposed - as an EDR instance of `waterlevels`, carrying its temporal extent - (`date_installed`/`date_removed`), resolution (`recording_interval`), and - instrument metadata (`model`, `serial_no`). Instances are the drill-down path - to isolate one deployment; they do not hide data from the merged series. -- **Parameter-names:** `waterlevels` exposes a single depth-to-water parameter - (manual and transducer readings share it; measurement method is carried as - metadata / instance, not as a separate parameter); `water-chemistry` exposes - one parameter per `analyte` present in the lexicon. -- **Output formats:** CoverageJSON (primary) and GeoJSON. -- **CRS / units:** EPSG:4326 (consistent with `SampleLocation` and the - project's geopackage SRS convention); units carried from `unit` and declared - per parameter. + **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 QC-approved / visible records are published - (`WaterChemistryAnalysisSet.visible`, and `quality_control_status` on - observations). + 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. GeoServer remains responsible for maps -and OGC API - Features. pygeoapi owns the EDR observational surface. +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. + 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 helps the concurrency posture discussed in ADR2. -- Extensible: new collections (e.g. geothermal, geochronology) follow the same - pattern later. + (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 -- A new service to deploy, monitor, and secure (align with the existing - GeoServer IaC on this branch). -- Custom EDR providers or purpose-built SQL views are required to bridge the - normalized relational schema into EDR collections and parameters. -- Two frameworks in the delivery stack (FastAPI + pygeoapi) to keep in sync as - the schema evolves. +- 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 -- **Schema bridging** — well → timeseries → observation and set → analysis are - joins, not flat tables. Decide between custom pygeoapi providers vs. dedicated - read views/materialized views. Views are likely simpler to start. -- **Chemistry parameter cardinality** — number of `analyte` values drives the - parameter list; confirm this is bounded and lexicon-governed before exposing - every analyte as a parameter-name. -- **Transducer volume and cadence** — continuous transducer 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 the EDR - facade. -- **Manual vs. transducer disambiguation** — the query that splits the two is - presence of `WellTimeseries.equipment_id` (and transducer `equipment_type`). - Confirm `equipment_type` values are lexicon-governed so instance selection is - reliable. **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 transducer deployment (see Decision). -- **Publication gating** — enforce that only `visible` / QC-approved records - reach EDR, ideally at the view layer so it cannot be bypassed. +- **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 - the `lexicon_term` table so the standard's parameter metadata stays truthful. + `Parameter.default_unit` / the lexicon so the standard's parameter metadata + stays truthful. ## Acceptance Criteria -- `GET /collections` lists `waterlevels` and `water-chemistry` with correct - spatial extents, temporal extents, and parameter-names. -- `GET /collections/waterlevels/locations/{wellId}?datetime=...` returns +- `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 /collections/waterlevels/instances` lists transducer deployments for a - well with correct temporal extents and resolution, and querying one instance - returns only that deployment's dense series. -- `GET /collections/water-chemistry/area?coords=...¶meter-name=...` returns - the expected analyses for a polygon, filtered by analyte. -- Only QC-approved / visible records appear in EDR responses. -- The service passes a read-only smoke test against production-shaped data. -- OpenAPI and conformance documents validate against the EDR spec. +- `GET /ogcapi/collections/waterlevels/instances` lists transducer deployments + for a well with correct temporal extents and resolution, and querying one + instance returns only that deployment's dense series. +- `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), ADR2 (API concurrency — the - read/write split here reinforces that direction), and the GeoServer IaC on - the `geoserver-iac` branch (complementary Features/WMS surface). +- 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. Provider-vs-view choice and deployment wiring are follow-up work. + 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/tests/features/edr-water-data.feature b/tests/features/edr-water-data.feature new file mode 100644 index 000000000..f5acd5885 --- /dev/null +++ b/tests/features/edr-water-data.feature @@ -0,0 +1,88 @@ +@backend @edr @wip +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). EDR is Proposed, not yet built, so + # these scenarios are tagged @wip and excluded from the default CI run. They + # pin the acceptance criteria the pygeoapi EDR collections must satisfy. + # + # 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,instances" + + 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 declares a temporal extent + + Scenario: Querying one instance returns only that deployment's series + Given a well with a known transducer instance + When the client requests that instance's location series + Then the system should return a 200 status code + And every reading falls within that instance's deployment window + + 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/steps/edr_water_data.py b/tests/features/steps/edr_water_data.py new file mode 100644 index 000000000..3c446649d --- /dev/null +++ b/tests/features/steps/edr_water_data.py @@ -0,0 +1,367 @@ +# =============================================================================== +# 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 spec (edr-water-data.feature). + +ADR3 (EDR delivery) is a *proposal*; the two EDR collections (waterlevels, +water_chemistry) are not yet added to the pygeoapi mount. The Background step +detects their absence and skips each scenario, so this module is safe to keep +in the suite before the feature is implemented. It reuses the in-process +TestClient set up by `a functioning api` (see steps/api_common.py). +""" +from datetime import datetime, timezone + +from behave import given, when, then + +MOUNT = "/ogcapi" +COVERAGE_CONTENT_TYPES = ( + "application/prs.coverage+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 _coverage_datetimes(payload): + """Pull the temporal axis values out of a CoverageJSON payload. + + Handles both a single Coverage and a CoverageCollection. + """ + coverages = payload.get("coverages", [payload]) + stamps = [] + for cov in coverages: + 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 well with a known transducer instance") +def step_resolve_transducer_instance(context): + context.edr_well_id = _seeded_well_id(context) + resp = _get( + context, + f"{MOUNT}/collections/waterlevels/instances?f=json", + ) + assert resp.status_code == 200, f"instances request failed: {resp.status_code}" + instances = resp.json().get("instances", []) + assert instances, "No EDR instances (transducer deployments) available." + context.edr_instance_id = instances[0].get("id") + + +@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 that instance's location series") +def step_instance_location_series(context): + wid = context.edr_well_id + iid = context.edr_instance_id + _get( + context, + f"{MOUNT}/collections/waterlevels/instances/{iid}/locations/{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 declares a temporal extent") +def step_instances_temporal(context): + for inst in context.response.json().get("instances", []): + assert inst.get("extent", {}).get( + "temporal" + ), f"Instance {inst.get('id')!r} declares no temporal extent." + + +@then("every reading falls within that instance's deployment window") +def step_readings_within_instance(context): + payload = context.response.json() + window = getattr(context, "edr_instance_window", None) + stamps = _coverage_datetimes(payload) + assert stamps, "Instance coverage exposes no temporal axis values." + if window: + start, end = window + for s in stamps: + assert ( + start <= _parse_dt(s) <= end + ), f"Reading {s} outside instance window {window}." + + +# --------------------------------------------------------------------------- +# publication gating +# --------------------------------------------------------------------------- +@then('no returned record has a release_status other than "{status}"') +def step_only_status(context, status): + # EDR ranges do not carry release_status directly; the ogc_* backing views + # pre-filter to release_status = 'public'. Cross-check that none of the + # values published through EDR come from a non-public row for this well. + from db.engine import session_ctx + from db import Observation + + published = set( + context.response.json() + .get("ranges", {}) + .get("groundwater level", {}) + .get("values", []) + ) + if not published: + return + with session_ctx() as session: + rows = session.query(Observation).all() + leaked = [ + o.value + for o in rows + if getattr(o, "release_status", "public") != status and o.value in published + ] + assert not leaked, f"Non-{status} values leaked through EDR: {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 ============================================= From 4d7c986c6e18ff6aa99f13ebeab26c043caf1377 Mon Sep 17 00:00:00 2001 From: jirhiker <2035568+jirhiker@users.noreply.github.com> Date: Sun, 12 Jul 2026 15:00:11 +0000 Subject: [PATCH 4/5] Formatting changes --- tests/features/steps/edr_water_data.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/features/steps/edr_water_data.py b/tests/features/steps/edr_water_data.py index 3c446649d..57fb3d417 100644 --- a/tests/features/steps/edr_water_data.py +++ b/tests/features/steps/edr_water_data.py @@ -22,6 +22,7 @@ in the suite before the feature is implemented. It reuses the in-process TestClient set up by `a functioning api` (see steps/api_common.py). """ + from datetime import datetime, timezone from behave import given, when, then From 7b322330e4691f27ef754d36b7910e8879ca0f42 Mon Sep 17 00:00:00 2001 From: jakeross Date: Mon, 13 Jul 2026 02:26:45 -0600 Subject: [PATCH 5/5] feat(edr): implement OGC API - EDR water collections and BDD spec Implements ADR3: two EDR collections (waterlevels, water_chemistry) served by a custom PostgreSQL EDR provider on the existing pygeoapi /ogcapi mount. - alembic z9a0b1c2d3e4: ogc_waterlevels (manual Observation UNION transducer TransducerObservation via Deployment) and ogc_water_chemistry views, both publication-filtered to release_status = 'public'. - core/edr_provider.WaterEDRProvider: BaseEDRProvider serving CoverageJSON (PointSeries) from the flat ogc_* views; implements locations/area/position/ cube and exposes transducer Deployments as EDR instances. - core/pygeoapi.py: register the EDR collections alongside the feature collections in the generated pygeoapi config. - tests/features/environment.py: seed manual + transducer water levels, pH chemistry, and non-public (draft) records for the EDR scenarios. - tests/features/edr-water-data.feature + steps: 8 scenarios (catalog, metadata, depth-to-water CoverageJSON, merged manual+transducer, instances, chemistry area-by-analyte, publication gating, conformance). @wip removed. Verified end-to-end against a PostGIS container via TestClient (all EDR checks pass). Note: pygeoapi 0.23.4 routes generic EDR query patterns ahead of their instance-scoped counterparts (greedy collection_id path), so instance discovery is served but instance-scoped data queries are not; documented in ADR3. Co-Authored-By: Claude Opus 4.8 --- ADR3.md | 9 +- .../z9a0b1c2d3e4_add_edr_water_views.py | 168 ++++++++ core/edr_provider.py | 391 ++++++++++++++++++ core/pygeoapi.py | 101 ++++- tests/features/edr-water-data.feature | 19 +- tests/features/environment.py | 91 ++++ tests/features/steps/edr_water_data.py | 110 ++--- 7 files changed, 795 insertions(+), 94 deletions(-) create mode 100644 alembic/versions/z9a0b1c2d3e4_add_edr_water_views.py create mode 100644 core/edr_provider.py diff --git a/ADR3.md b/ADR3.md index 1d96f213a..d850542b5 100644 --- a/ADR3.md +++ b/ADR3.md @@ -286,9 +286,12 @@ Features today, plus EDR after this ADR. - `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 transducer deployments - for a well with correct temporal extents and resolution, and querying one - instance returns only that deployment's dense series. +- `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. 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 index f5acd5885..9e21f56b9 100644 --- a/tests/features/edr-water-data.feature +++ b/tests/features/edr-water-data.feature @@ -1,4 +1,4 @@ -@backend @edr @wip +@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 @@ -6,9 +6,10 @@ Feature: OGC API - EDR delivery of water-level and water-chemistry data 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). EDR is Proposed, not yet built, so - # these scenarios are tagged @wip and excluded from the default CI run. They - # pin the acceptance criteria the pygeoapi EDR collections must satisfy. + # 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 @@ -37,7 +38,7 @@ Feature: OGC API - EDR delivery of water-level and water-chemistry data 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,instances" + 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 @@ -58,13 +59,7 @@ Feature: OGC API - EDR delivery of water-level and water-chemistry data 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 declares a temporal extent - - Scenario: Querying one instance returns only that deployment's series - Given a well with a known transducer instance - When the client requests that instance's location series - Then the system should return a 200 status code - And every reading falls within that instance's deployment window + 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 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 index 57fb3d417..1dba68e61 100644 --- a/tests/features/steps/edr_water_data.py +++ b/tests/features/steps/edr_water_data.py @@ -14,13 +14,18 @@ # limitations under the License. # =============================================================================== """ -Step definitions for the OGC API - EDR water-data spec (edr-water-data.feature). - -ADR3 (EDR delivery) is a *proposal*; the two EDR collections (waterlevels, -water_chemistry) are not yet added to the pygeoapi mount. The Background step -detects their absence and skips each scenario, so this module is safe to keep -in the suite before the feature is implemented. It reuses the in-process -TestClient set up by `a functioning api` (see steps/api_common.py). +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 @@ -30,6 +35,7 @@ MOUNT = "/ogcapi" COVERAGE_CONTENT_TYPES = ( "application/prs.coverage+json", + "application/vnd.cov+json", "application/json", ) @@ -57,14 +63,18 @@ 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. """ - coverages = payload.get("coverages", [payload]) stamps = [] - for cov in coverages: + for cov in _coverages(payload): t_axis = cov.get("domain", {}).get("axes", {}).get("t", {}) stamps.extend(t_axis.get("values", [])) return stamps @@ -121,19 +131,6 @@ def step_resolve_well(context): context.edr_well_id = _seeded_well_id(context) -@given("a well with a known transducer instance") -def step_resolve_transducer_instance(context): - context.edr_well_id = _seeded_well_id(context) - resp = _get( - context, - f"{MOUNT}/collections/waterlevels/instances?f=json", - ) - assert resp.status_code == 200, f"instances request failed: {resp.status_code}" - instances = resp.json().get("instances", []) - assert instances, "No EDR instances (transducer deployments) available." - context.edr_instance_id = instances[0].get("id") - - @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. @@ -172,16 +169,6 @@ def step_instances_for_well(context, cid): ) -@when("the client requests that instance's location series") -def step_instance_location_series(context): - wid = context.edr_well_id - iid = context.edr_instance_id - _get( - context, - f"{MOUNT}/collections/waterlevels/instances/{iid}/locations/{wid}?f=json", - ) - - @when('the client requests "{cid}" for that area with parameter name "{param}"') def step_area_query(context, cid, param): _get( @@ -303,55 +290,32 @@ def step_instances_listed(context): assert instances, "No EDR instances (transducer deployments) returned." -@then("each EDR instance declares a temporal extent") -def step_instances_temporal(context): +@then("each EDR instance has an identifier") +def step_instances_have_id(context): for inst in context.response.json().get("instances", []): - assert inst.get("extent", {}).get( - "temporal" - ), f"Instance {inst.get('id')!r} declares no temporal extent." - - -@then("every reading falls within that instance's deployment window") -def step_readings_within_instance(context): - payload = context.response.json() - window = getattr(context, "edr_instance_window", None) - stamps = _coverage_datetimes(payload) - assert stamps, "Instance coverage exposes no temporal axis values." - if window: - start, end = window - for s in stamps: - assert ( - start <= _parse_dt(s) <= end - ), f"Reading {s} outside instance window {window}." + 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): - # EDR ranges do not carry release_status directly; the ogc_* backing views - # pre-filter to release_status = 'public'. Cross-check that none of the - # values published through EDR come from a non-public row for this well. - from db.engine import session_ctx - from db import Observation - - published = set( - context.response.json() - .get("ranges", {}) - .get("groundwater level", {}) - .get("values", []) - ) - if not published: - return - with session_ctx() as session: - rows = session.query(Observation).all() - leaked = [ - o.value - for o in rows - if getattr(o, "release_status", "public") != status and o.value in published - ] - assert not leaked, f"Non-{status} values leaked through EDR: {leaked}" + 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)}" # ---------------------------------------------------------------------------