diff --git a/alembic/versions/b2c3d4e5f6a7_transducer_data_maturity.py b/alembic/versions/b2c3d4e5f6a7_transducer_data_maturity.py new file mode 100644 index 000000000..3037e3cd0 --- /dev/null +++ b/alembic/versions/b2c3d4e5f6a7_transducer_data_maturity.py @@ -0,0 +1,122 @@ +"""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 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 +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", + ) + + # 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( + "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..36052df1c 100644 --- a/docs/automated-ingestion-pipeline-plan.md +++ b/docs/automated-ingestion-pipeline-plan.md @@ -277,13 +277,26 @@ 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 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. + +⬜ 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 =============================================