feat(transducer): add data_maturity to observations

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.

automated_ingestion/ocotillo/loader.py

@@ -71,13 +71,24 @@ 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],
     deployment_id: int,
     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()

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"

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."
     }
   ]
 }

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
     )

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
 

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):

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 =============================================

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 =============================================