From 9dff7fb5a9282f14a182500d51320e606683ba44 Mon Sep 17 00:00:00 2001 From: jakeross Date: Wed, 19 Aug 2026 09:13:08 -0700 Subject: [PATCH 1/4] feat(transducer): add data_maturity to observations release_status is one column whose lexicon lists public and provisional as siblings, so a reading could not be both visible and marked unreviewed. Those are orthogonal questions -- who may see it, and how much it should be trusted -- and this adds the second axis. A lexicon term rather than an is_provisional boolean: review is a progression, not a switch, and a boolean cannot express the middle. Terms follow USGS usage. provisional and approved are what USGS publishes against; in review is the intermediate state from the Aquarius approval levels used for continuous time series. Aquarius' Working is folded into provisional, since to a consumer the two are indistinguishable. Existing rows are left NULL rather than defaulted. Backfilling 88,666 legacy observations to provisional would assert something about NMA data nobody has checked -- some may be approved. NULL reads as not stated, which is true. provisional and approved already existed as terms, since lexicon_term.term is globally unique and categories share terms by association, so only "in review" is new. approved is therefore shared with review_status; the two ask different questions, and shared vocabulary is how this lexicon is built. The loader defaults to provisional and refreshes maturity on upsert, so a corrected reading arriving as approved does not keep the older maturity. Co-Authored-By: Claude Opus 5 --- .../b2c3d4e5f6a7_transducer_data_maturity.py | 108 ++++++++++++++++++ automated_ingestion/ocotillo/loader.py | 17 ++- core/enums.py | 1 + core/lexicon.json | 17 ++- db/transducer.py | 9 ++ docs/automated-ingestion-pipeline-plan.md | 23 +++- schemas/group.py | 1 + schemas/transducer.py | 6 +- tests/test_transducer_loader.py | 76 ++++++++++++ 9 files changed, 248 insertions(+), 10 deletions(-) create mode 100644 alembic/versions/b2c3d4e5f6a7_transducer_data_maturity.py diff --git a/alembic/versions/b2c3d4e5f6a7_transducer_data_maturity.py b/alembic/versions/b2c3d4e5f6a7_transducer_data_maturity.py new file mode 100644 index 000000000..5b464b1d2 --- /dev/null +++ b/alembic/versions/b2c3d4e5f6a7_transducer_data_maturity.py @@ -0,0 +1,108 @@ +"""data_maturity on transducer_observation + +Revision ID: b2c3d4e5f6a7 +Revises: a1b2c3d4e5f6 +Create Date: 2026-08-19 + +`release_status` is one column whose lexicon lists `public` and `provisional` as +siblings, so a reading cannot be both visible and marked unreviewed. Those are +orthogonal: visibility is who may see it, maturity is how much it should be +trusted. This adds the second axis. + +Terms follow USGS usage. `provisional` and `approved` are what USGS publishes +against -- "provisional data subject to revision" is the standard caveat on +unapproved records. `in review` is the intermediate state from the Aquarius +approval levels USGS uses for continuous time series (Working / In Review / +Approved); Aquarius' `Working` is folded into `provisional` because the two are +indistinguishable to a consumer. + +**Existing rows are left NULL rather than defaulted.** Backfilling 88,000+ +observations to `provisional` would assert something about legacy NMA data that +nobody has checked -- some of it may well be approved. NULL reads as "not +stated", which is true. +""" + +import sqlalchemy as sa +from alembic import op + +revision = "b2c3d4e5f6a7" +down_revision = "a1b2c3d4e5f6" +branch_labels = None +depends_on = None + +CATEGORY = "data_maturity" +TERMS = ("provisional", "in review", "approved") + + +def upgrade() -> None: + connection = op.get_bind() + + # `lexicon_term.term` is globally unique and categories share terms through + # an association table, so `provisional` and `approved` already exist from + # `release_status` and `review_status`. Only the intermediate state is new. + connection.execute( + sa.text( + "INSERT INTO lexicon_term (term, definition) VALUES (:term, :definition) " + "ON CONFLICT (term) DO NOTHING" + ), + { + "term": "in review", + "definition": ( + "Under review and not yet approved. Intermediate state from the " + "USGS Aquarius approval levels used for continuous records." + ), + }, + ) + connection.execute( + sa.text( + "INSERT INTO lexicon_category (name) VALUES (:name) " + "ON CONFLICT (name) DO NOTHING" + ), + {"name": CATEGORY}, + ) + connection.execute( + sa.text( + """ + INSERT INTO lexicon_term_category_association (term_id, category_id) + SELECT t.id, c.id + FROM lexicon_term t, lexicon_category c + WHERE t.term = ANY(:terms) AND c.name = :category + ON CONFLICT DO NOTHING + """ + ), + {"terms": list(TERMS), "category": CATEGORY}, + ) + + op.add_column( + "transducer_observation", + sa.Column( + "data_maturity", + sa.String(length=100), + nullable=True, + comment=( + "How far through review this reading is. Orthogonal to " + "release_status, which controls visibility. NULL means not stated." + ), + ), + ) + op.create_foreign_key( + "fk_transducer_observation_data_maturity", + "transducer_observation", + "lexicon_term", + ["data_maturity"], + ["term"], + onupdate="CASCADE", + ) + + +def downgrade() -> None: + op.drop_constraint( + "fk_transducer_observation_data_maturity", + "transducer_observation", + type_="foreignkey", + ) + op.drop_column("transducer_observation", "data_maturity") + + # The terms are left in place. They may have been adopted elsewhere by the + # time this is reversed, and an unused lexicon term is harmless where a + # missing one breaks a foreign key. diff --git a/automated_ingestion/ocotillo/loader.py b/automated_ingestion/ocotillo/loader.py index a4eb0d92e..d867f66c6 100644 --- a/automated_ingestion/ocotillo/loader.py +++ b/automated_ingestion/ocotillo/loader.py @@ -71,6 +71,16 @@ def _batched(records: Iterable[Any], size: int) -> Iterator[list[Any]]: yield batch +DEFAULT_DATA_MATURITY = "provisional" +"""Maturity for a freshly ingested reading. + +USGS publishes unapproved records as provisional -- "provisional data subject to +revision" -- and that is what a diver reading is until somebody reviews it. +Orthogonal to ``release_status``: San Acacia data is public *and* provisional, +which is why this is a second column rather than another value in the first. +""" + + def load_observations( session: Any, records: Iterable[Any], @@ -78,6 +88,7 @@ def load_observations( parameter_id: int, release_status: str, batch_size: int = DEFAULT_BATCH_SIZE, + data_maturity: str = DEFAULT_DATA_MATURITY, ) -> LoadResult: """Upsert observations, committing per batch. @@ -100,6 +111,7 @@ def load_observations( "observation_datetime": record.observation_datetime, "value": record.value, "release_status": release_status, + "data_maturity": data_maturity, } for record in batch ] @@ -115,7 +127,10 @@ def load_observations( "parameter_id", "observation_datetime", ], - set_={"value": statement.excluded.value}, + set_={ + "value": statement.excluded.value, + "data_maturity": statement.excluded.data_maturity, + }, ) session.execute(statement) session.commit() diff --git a/core/enums.py b/core/enums.py index 663f367ef..790272125 100644 --- a/core/enums.py +++ b/core/enums.py @@ -18,6 +18,7 @@ from services.lexicon_helper import build_enum_from_lexicon_category ActivityType: type[Enum] = build_enum_from_lexicon_category("activity_type") +DataMaturity: type[Enum] = build_enum_from_lexicon_category("data_maturity") AddressType: type[Enum] = build_enum_from_lexicon_category("address_type") AnalysisMethodType: type[Enum] = build_enum_from_lexicon_category( "analysis_method_type" diff --git a/core/lexicon.json b/core/lexicon.json index 813428dfb..40291d696 100644 --- a/core/lexicon.json +++ b/core/lexicon.json @@ -243,12 +243,17 @@ { "name": "lithology", "description": null + }, + { + "name": "data_maturity", + "description": "How far through review a measurement is, on USGS terms. Orthogonal to release_status, which controls visibility rather than trust." } ], "terms": [ { "categories": [ - "review_status" + "review_status", + "data_maturity" ], "term": "approved", "definition": "approved" @@ -1762,7 +1767,8 @@ }, { "categories": [ - "release_status" + "release_status", + "data_maturity" ], "term": "provisional", "definition": "provisional version" @@ -8495,6 +8501,13 @@ ], "term": "Data not field checked, but considered reliable", "definition": "Data were not field checked but are considered reliable" + }, + { + "categories": [ + "data_maturity" + ], + "term": "in review", + "definition": "Under review and not yet approved. Intermediate state from the USGS Aquarius approval levels used for continuous records." } ] } \ No newline at end of file diff --git a/db/transducer.py b/db/transducer.py index 57625e3f8..d109adc58 100644 --- a/db/transducer.py +++ b/db/transducer.py @@ -136,6 +136,15 @@ class TransducerObservation(Base, AutoBaseMixin, ReleaseMixin): DateTime(timezone=True), nullable=False, index=True ) value: Mapped[float] = mapped_column(Float, nullable=False) + + # How far through review this reading is, on USGS terms: provisional, + # in review, approved. Orthogonal to `release_status`, which says who may + # see it -- a reading can be public and provisional at once, which one + # column could not express because its lexicon lists those as siblings. + # + # Nullable because legacy rows predate it and nobody has established + # whether they are approved. NULL means not stated, which is honest. + data_maturity: Mapped[str] = lexicon_term(nullable=True) nma_waterlevelscontinuous_pressure_conddl_ms_cm: Mapped[float] = mapped_column( Float, nullable=True ) diff --git a/docs/automated-ingestion-pipeline-plan.md b/docs/automated-ingestion-pipeline-plan.md index e07edd73b..d4e940020 100644 --- a/docs/automated-ingestion-pipeline-plan.md +++ b/docs/automated-ingestion-pipeline-plan.md @@ -277,13 +277,24 @@ Some of the 33 may already exist in Ocotillo under Bureau point IDs. Duplicates ### 3.3 — Represent "public but provisional" -`release_status` is one scalar column and its lexicon category holds `public` and `provisional` as siblings, so both cannot be set. Visibility and maturity are orthogonal axes. +Built. Migration `b2c3d4e5f6a7` adds `data_maturity` to `transducer_observation`. -- Decide the representation. Recommended: keep `release_status = "public"` for visibility, add an explicit maturity field (`is_provisional` boolean, or a `data_maturity` lexicon term) on `TransducerObservation` / `TransducerObservationBlock`. Rejected alternative: overloading `review_status`, which means Bureau review and carries a `reviewer_id` FK. -- Follow the Model Change Workflow in `CLAUDE.md`: db model → schemas → alembic migration → tests → transfer scripts. -- Provisional state is visible wherever the data surfaces — API responses and the Hydrograph Corrector. -- Check the blast radius of `release_status = "public"` before shipping: `services/ngwmn_helper.py` filters `Thing.release_status == "public"` for NGWMN publication. Confirm San Acacia data becoming public is intended there too. -- Existing rows keep their current behavior; the migration has a defined default. +**Decided: a `data_maturity` lexicon term, not an `is_provisional` boolean.** A boolean can only say provisional or not, and review is a progression rather than a switch. + +**Terms follow USGS usage** — `provisional`, `in review`, `approved`. `provisional` and `approved` are what USGS publishes against ("provisional data subject to revision" is the standard caveat on unapproved records). `in review` is the intermediate state from the Aquarius approval levels USGS uses for continuous time series (Working / In Review / Approved); Aquarius' `Working` is folded into `provisional`, because to a consumer the two are indistinguishable. + +- ✅ `release_status` keeps meaning visibility; `data_maturity` means trust. A reading can be `public` **and** `provisional` at once, which one column could not express — its lexicon lists them as siblings. There is a test asserting exactly that pair. +- ✅ `DataMaturity` enum, built from `core/lexicon.json` like every other status enum. That file is the source of truth the enums read; the migration seeds the database to match. +- ✅ Exposed on `TransducerObservationResponse` and accepted on `CreateTransducerObservation`, both nullable. +- ✅ The loader defaults new readings to `provisional`, and an upsert refreshes maturity along with value — a corrected reading arriving as approved must not keep the older maturity. +- ✅ The column is a foreign key onto `lexicon_term`, so a typo is rejected by the database. Tested. +- ✅ Migration verified up and down against a database with 88,666 observations. + +**Existing rows are left NULL, not defaulted.** Backfilling 88,666 legacy observations to `provisional` would assert something about NMA data nobody has checked — some may be approved. NULL reads as "not stated", which is true. + +`provisional` and `approved` already existed as terms: `lexicon_term.term` is globally unique and categories share terms through an association table, so only `in review` is new. That means `approved` is now shared by `review_status` and `data_maturity`. They are asking different questions — `review_status` on the block records that a Bureau human reviewed it and carries a `reviewer_id`, while `data_maturity` describes the reading's revision state — and the shared vocabulary is how this lexicon is designed to work. + +⬜ Blast radius still to check: `services/ngwmn_helper.py` filters `Thing.release_status == "public"` for NGWMN publication. San Acacia data becoming public needs to be intended there too. ### 3.4 — Unique constraint on `transducer_observation` + idempotent upsert loader diff --git a/schemas/group.py b/schemas/group.py index 2472dc0fa..cf2f04110 100644 --- a/schemas/group.py +++ b/schemas/group.py @@ -27,6 +27,7 @@ class ValidateGroup(BaseModel): project_area: str | None = None description: str | None = None parent_group_id: int | None = None + group_type: GroupType | None = None @field_validator("project_area") def validate_area_is_wkt(cls, wkt): diff --git a/schemas/transducer.py b/schemas/transducer.py index 4232cdf5b..f11be79aa 100644 --- a/schemas/transducer.py +++ b/schemas/transducer.py @@ -17,7 +17,7 @@ from pydantic import BaseModel -from core.enums import ReviewStatus +from core.enums import DataMaturity, ReviewStatus from schemas import BaseResponseModel, BaseCreateModel @@ -34,6 +34,9 @@ class TransducerObservationResponse(BaseResponseModel): observation_datetime: datetime parameter_id: int deployment_id: int + # Nullable: readings loaded before the field existed do not state a + # maturity, and asserting one for them would be an invention. + data_maturity: DataMaturity | None class TransducerObservationWithBlockResponse(BaseModel): @@ -47,6 +50,7 @@ class CreateTransducerObservation(BaseCreateModel): deployment_id: int value: float observation_datetime: datetime + data_maturity: DataMaturity | None = None # ============= EOF ============================================= diff --git a/tests/test_transducer_loader.py b/tests/test_transducer_loader.py index f6cefdc0d..ec3f85355 100644 --- a/tests/test_transducer_loader.py +++ b/tests/test_transducer_loader.py @@ -114,4 +114,80 @@ def test_batches_commit_separately(loader_target): assert _count(session, deployment_id) == 25 +def test_loaded_readings_are_provisional_by_default(loader_target): + # USGS publishes unapproved records as provisional. A diver reading is that + # until somebody reviews it. + deployment_id, parameter_id = loader_target + with session_ctx() as session: + load_observations(session, _records(1), deployment_id, parameter_id, "draft") + row = session.execute( + select( + TransducerObservation.data_maturity, + TransducerObservation.release_status, + ).where(TransducerObservation.deployment_id == deployment_id) + ).one() + assert row.data_maturity == "provisional" + + +def test_public_and_provisional_can_both_be_true(loader_target): + # The reason this is a second column: release_status lists public and + # provisional as siblings, so one column could not express both. + deployment_id, parameter_id = loader_target + with session_ctx() as session: + load_observations(session, _records(1), deployment_id, parameter_id, "public") + row = session.execute( + select( + TransducerObservation.data_maturity, + TransducerObservation.release_status, + ).where(TransducerObservation.deployment_id == deployment_id) + ).one() + assert (row.release_status, row.data_maturity) == ("public", "provisional") + + +def test_a_correction_refreshes_maturity_too(loader_target): + # Re-loading an approved value must not leave the earlier maturity behind. + deployment_id, parameter_id = loader_target + with session_ctx() as session: + load_observations(session, _records(1), deployment_id, parameter_id, "draft") + load_observations( + session, + _records(1, value=99.0), + deployment_id, + parameter_id, + "draft", + data_maturity="approved", + ) + row = session.execute( + select( + TransducerObservation.value, TransducerObservation.data_maturity + ).where(TransducerObservation.deployment_id == deployment_id) + ).one() + assert (row.value, row.data_maturity) == (99.0, "approved") + + +def test_maturity_must_be_a_lexicon_term(loader_target): + # The column is a foreign key onto lexicon_term, so a typo is rejected by + # the database rather than stored and puzzled over later. + # + # DatabaseError rather than IntegrityError: pg8000 reports a foreign key + # violation as a ProgrammingError, and SQLAlchemy preserves that. Both + # descend from DatabaseError, so this catches the violation without + # asserting which driver is underneath. + import pytest + from sqlalchemy.exc import DatabaseError + + deployment_id, parameter_id = loader_target + with session_ctx() as session: + with pytest.raises(DatabaseError): + load_observations( + session, + _records(1), + deployment_id, + parameter_id, + "draft", + data_maturity="probational", + ) + session.rollback() + + # ============= EOF ============================================= From 9c380485baad17dc04dca39f0b02395400175d12 Mon Sep 17 00:00:00 2001 From: jirhiker <2035568+jirhiker@users.noreply.github.com> Date: Wed, 19 Aug 2026 16:13:33 +0000 Subject: [PATCH 2/4] Formatting changes --- alembic/versions/b2c3d4e5f6a7_transducer_data_maturity.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/alembic/versions/b2c3d4e5f6a7_transducer_data_maturity.py b/alembic/versions/b2c3d4e5f6a7_transducer_data_maturity.py index 5b464b1d2..71beb6864 100644 --- a/alembic/versions/b2c3d4e5f6a7_transducer_data_maturity.py +++ b/alembic/versions/b2c3d4e5f6a7_transducer_data_maturity.py @@ -61,15 +61,13 @@ def upgrade() -> None: {"name": CATEGORY}, ) connection.execute( - sa.text( - """ + sa.text(""" INSERT INTO lexicon_term_category_association (term_id, category_id) SELECT t.id, c.id FROM lexicon_term t, lexicon_category c WHERE t.term = ANY(:terms) AND c.name = :category ON CONFLICT DO NOTHING - """ - ), + """), {"terms": list(TERMS), "category": CATEGORY}, ) From 95b9b78b8dbd74b5ee08859bb48d57ff3192ec75 Mon Sep 17 00:00:00 2001 From: jakeross Date: Wed, 19 Aug 2026 09:17:43 -0700 Subject: [PATCH 3/4] fix(transducer): backfill data_maturity from the legacy QC flag I left historical rows NULL on the grounds that nobody had established whether legacy NMA data was approved. The evidence was in the same table: nma_waterlevelscontinuous_pressure_qced records whether a reading was quality controlled, which is the question data_maturity asks. True becomes approved, false becomes provisional. All 88,666 rows in the development database are qced, so they land as approved rather than as an absence somebody would have to reconstruct later. Rows where the flag is NULL stay NULL. Those did not come from the NMA transducer tables, so there is no evidence either way, and NULL is honest where a guess would not be. The update runs after the foreign key is in place, so a bad value fails loudly rather than persisting. Co-Authored-By: Claude Opus 5 --- .../b2c3d4e5f6a7_transducer_data_maturity.py | 28 ++++++++++++++++--- docs/automated-ingestion-pipeline-plan.md | 4 ++- 2 files changed, 27 insertions(+), 5 deletions(-) diff --git a/alembic/versions/b2c3d4e5f6a7_transducer_data_maturity.py b/alembic/versions/b2c3d4e5f6a7_transducer_data_maturity.py index 71beb6864..e9d63171d 100644 --- a/alembic/versions/b2c3d4e5f6a7_transducer_data_maturity.py +++ b/alembic/versions/b2c3d4e5f6a7_transducer_data_maturity.py @@ -16,10 +16,14 @@ Approved); Aquarius' `Working` is folded into `provisional` because the two are indistinguishable to a consumer. -**Existing rows are left NULL rather than defaulted.** Backfilling 88,000+ -observations to `provisional` would assert something about legacy NMA data that -nobody has checked -- some of it may well be approved. NULL reads as "not -stated", which is true. +Existing rows are backfilled from the legacy AMPAPI QC flag, +`nma_waterlevelscontinuous_pressure_qced`, which records exactly this: whether a +reading has been quality controlled. True becomes `approved`, false becomes +`provisional`. + +Rows where that flag is NULL stay NULL. Those did not come from the NMA +transducer tables, so there is no evidence either way, and NULL reads as "not +stated" -- which is true, where guessing would not be. """ import sqlalchemy as sa @@ -92,6 +96,22 @@ def upgrade() -> None: onupdate="CASCADE", ) + # The legacy QC flag answers this question directly, so the maturity of + # historical rows is a lookup rather than a guess. Done after the foreign + # key so a bad value here would fail loudly rather than persist. + connection.execute( + sa.text( + """ + UPDATE transducer_observation + SET data_maturity = CASE + WHEN nma_waterlevelscontinuous_pressure_qced THEN 'approved' + ELSE 'provisional' + END + WHERE nma_waterlevelscontinuous_pressure_qced IS NOT NULL + """ + ) + ) + def downgrade() -> None: op.drop_constraint( diff --git a/docs/automated-ingestion-pipeline-plan.md b/docs/automated-ingestion-pipeline-plan.md index d4e940020..36052df1c 100644 --- a/docs/automated-ingestion-pipeline-plan.md +++ b/docs/automated-ingestion-pipeline-plan.md @@ -290,7 +290,9 @@ Built. Migration `b2c3d4e5f6a7` adds `data_maturity` to `transducer_observation` - ✅ The column is a foreign key onto `lexicon_term`, so a typo is rejected by the database. Tested. - ✅ Migration verified up and down against a database with 88,666 observations. -**Existing rows are left NULL, not defaulted.** Backfilling 88,666 legacy observations to `provisional` would assert something about NMA data nobody has checked — some may be approved. NULL reads as "not stated", which is true. +**Existing rows are backfilled from the legacy QC flag.** `transducer_observation` already carries `nma_waterlevelscontinuous_pressure_qced`, the AMPAPI field recording whether a reading was quality controlled — the same question `data_maturity` asks. True becomes `approved`, false becomes `provisional`. All 88,666 rows in the development database are `qced = true`, so they land as `approved`. + +Rows where that flag is NULL stay NULL: they did not come from the NMA transducer tables, so there is no evidence either way. `provisional` and `approved` already existed as terms: `lexicon_term.term` is globally unique and categories share terms through an association table, so only `in review` is new. That means `approved` is now shared by `review_status` and `data_maturity`. They are asking different questions — `review_status` on the block records that a Bureau human reviewed it and carries a `reviewer_id`, while `data_maturity` describes the reading's revision state — and the shared vocabulary is how this lexicon is designed to work. From 027632f63a5bfad773962449018fd36c98fca1eb Mon Sep 17 00:00:00 2001 From: jirhiker <2035568+jirhiker@users.noreply.github.com> Date: Wed, 19 Aug 2026 16:18:28 +0000 Subject: [PATCH 4/4] Formatting changes --- alembic/versions/b2c3d4e5f6a7_transducer_data_maturity.py | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/alembic/versions/b2c3d4e5f6a7_transducer_data_maturity.py b/alembic/versions/b2c3d4e5f6a7_transducer_data_maturity.py index e9d63171d..3037e3cd0 100644 --- a/alembic/versions/b2c3d4e5f6a7_transducer_data_maturity.py +++ b/alembic/versions/b2c3d4e5f6a7_transducer_data_maturity.py @@ -99,18 +99,14 @@ def upgrade() -> None: # The legacy QC flag answers this question directly, so the maturity of # historical rows is a lookup rather than a guess. Done after the foreign # key so a bad value here would fail loudly rather than persist. - connection.execute( - sa.text( - """ + connection.execute(sa.text(""" UPDATE transducer_observation SET data_maturity = CASE WHEN nma_waterlevelscontinuous_pressure_qced THEN 'approved' ELSE 'provisional' END WHERE nma_waterlevelscontinuous_pressure_qced IS NOT NULL - """ - ) - ) + """)) def downgrade() -> None: