Merge pull request #868 from DataIntegrationGroup/feat/backfill-acoustic-data-maturity

feat(transducer): backfill data_maturity on acoustic (Wellntel) observations

Author: jirhiker

data_migrations/migrations/20260820_0001_backfill_acoustic_data_maturity.py

@@ -0,0 +1,88 @@
+# ===============================================================================
+# Copyright 2026 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.
+# ===============================================================================
+"""
+Set `data_maturity` on the acoustic (Wellntel) transducer observations that
+alembic revision `b2c3d4e5f6a7` left NULL.
+
+That revision backfilled maturity from `nma_waterlevelscontinuous_pressure_qced`,
+the AMPAPI flag recording whether a reading was quality controlled. Acoustic
+readings have no such flag -- AMPAPI's `WaterLevelsContinuous_Acoustic` table has
+no `QCed` column at all -- so all 394,086 of them were skipped, which is the
+entire acoustic record (BDMS-1169).
+
+`MATURITY` is a deliberate choice, not a derivation. There is no QC field in the
+acoustic legacy schema to read, so nothing here computes the answer; the value
+below is the one recorded for these readings, applied uniformly. The transfer's
+`review_status='approved'` blocks are *not* evidence for it -- those come from
+`PublicRelease`, which every acoustic source row carries and which describes
+visibility rather than review.
+
+Rows are matched on `nma_waterlevelscontinuous_acoustic_global_id`, the AMPAPI
+row identity. It is written by `WaterLevelsContinuousAcousticTransferer` on every
+acoustic row and never by the pressure transferer, so it is the provenance
+marker: 394,086 rows carry it, and they are exactly the rows with no
+`pressure_qced`.
+
+Only rows where `data_maturity` is already NULL are touched. Re-running is
+therefore a no-op, and a maturity set deliberately since -- by the hydrograph
+corrector, or by a later migration once the acoustic QC history is known -- is
+left alone rather than reset to the blanket value.
+"""
+
+from sqlalchemy import update
+from sqlalchemy.orm import Session
+
+from data_migrations.base import DataMigration
+from db.transducer import TransducerObservation
+
+MATURITY = "approved"
+
+
+def run(session: Session) -> None:
+    """Set the maturity on acoustic observations that have none."""
+    result = session.execute(
+        update(TransducerObservation)
+        .where(
+            TransducerObservation.nma_waterlevelscontinuous_acoustic_global_id.isnot(
+                None
+            ),
+            TransducerObservation.data_maturity.is_(None),
+        )
+        .values(data_maturity=MATURITY)
+        .execution_options(synchronize_session=False)
+    )
+    print(
+        f"  set data_maturity={MATURITY!r} on {result.rowcount} acoustic observations"
+    )
+    return None
+
+
+MIGRATION = DataMigration(
+    id="20260820_0001_backfill_acoustic_data_maturity",
+    alembic_revision="b2c3d4e5f6a7",
+    name="Backfill data_maturity on acoustic (Wellntel) observations",
+    description=(
+        "Revision b2c3d4e5f6a7 backfilled data_maturity from the pressure QC "
+        "flag, which acoustic readings do not have, leaving the entire 394,086 "
+        f"row Wellntel record NULL (BDMS-1169). Sets it to {MATURITY!r}. Only "
+        "touches rows whose maturity is still NULL."
+    ),
+    run=run,
+    is_repeatable=False,
+)
+
+
+# ============= EOF =============================================

tests/test_data_migrations.py

@@ -14,19 +14,25 @@
 # limitations under the License.
 # ===============================================================================
 import importlib
+from datetime import datetime, timedelta, timezone
 
-from sqlalchemy import select
+from sqlalchemy import delete, select
 
 move_notes = importlib.import_module(
     "data_migrations.migrations.20260205_0001_move_nma_location_notes"
 )
 publish_project_areas = importlib.import_module(
     "data_migrations.migrations.20260714_0001_publish_project_areas"
 )
+backfill_acoustic_maturity = importlib.import_module(
+    "data_migrations.migrations.20260820_0001_backfill_acoustic_data_maturity"
+)
 from db.location import Location
 from db.notes import Notes
 from db.group import Group
 from db.engine import session_ctx
+from db.transducer import TransducerObservation
+from tests import get_parameter_id
 
 
 def test_move_nma_location_notes_creates_notes_and_clears_field():
@@ -139,3 +145,91 @@ def test_publish_project_areas_marks_project_area_groups_public():
         session.delete(draft_with_area)
         session.delete(draft_without_area)
         session.commit()
+
+
+def test_backfill_acoustic_data_maturity_only_touches_null_acoustic_rows(
+    sensor_to_water_well_thing_deployment,
+):
+    deployment_id = sensor_to_water_well_thing_deployment.id
+    parameter_id = get_parameter_id("groundwater level", "Field Parameter")
+    observed = datetime(2019, 7, 23, 12, 0, tzinfo=timezone.utc)
+
+    with session_ctx() as session:
+        # An acoustic row with no maturity -- the case this migration exists for.
+        acoustic = TransducerObservation(
+            parameter_id=parameter_id,
+            deployment_id=deployment_id,
+            observation_datetime=observed,
+            value=42.0,
+            nma_waterlevelscontinuous_acoustic_global_id="ACOUSTIC-NULL",
+        )
+        # An acoustic row whose maturity was already set deliberately. The
+        # blanket value must not overwrite a decision someone made.
+        acoustic_already_set = TransducerObservation(
+            parameter_id=parameter_id,
+            deployment_id=deployment_id,
+            observation_datetime=observed + timedelta(hours=1),
+            value=43.0,
+            nma_waterlevelscontinuous_acoustic_global_id="ACOUSTIC-SET",
+            data_maturity="provisional",
+        )
+        # A pressure row with no maturity. NULL here means the pressure QC flag
+        # was NULL, which is a different question -- leave it alone.
+        pressure = TransducerObservation(
+            parameter_id=parameter_id,
+            deployment_id=deployment_id,
+            observation_datetime=observed + timedelta(hours=2),
+            value=44.0,
+            nma_waterlevelscontinuous_pressure_global_id="PRESSURE-NULL",
+        )
+        session.add_all([acoustic, acoustic_already_set, pressure])
+        session.commit()
+        ids = (acoustic.id, acoustic_already_set.id, pressure.id)
+
+        try:
+            backfill_acoustic_maturity.run(session)
+
+            session.refresh(acoustic)
+            session.refresh(acoustic_already_set)
+            session.refresh(pressure)
+            assert acoustic.data_maturity == backfill_acoustic_maturity.MATURITY
+            assert acoustic_already_set.data_maturity == "provisional"
+            assert pressure.data_maturity is None
+        finally:
+            session.execute(
+                delete(TransducerObservation).where(TransducerObservation.id.in_(ids))
+            )
+            session.commit()
+
+
+def test_backfill_acoustic_data_maturity_is_idempotent(
+    sensor_to_water_well_thing_deployment,
+):
+    deployment_id = sensor_to_water_well_thing_deployment.id
+    parameter_id = get_parameter_id("groundwater level", "Field Parameter")
+
+    with session_ctx() as session:
+        observation = TransducerObservation(
+            parameter_id=parameter_id,
+            deployment_id=deployment_id,
+            observation_datetime=datetime(2020, 1, 1, tzinfo=timezone.utc),
+            value=45.0,
+            nma_waterlevelscontinuous_acoustic_global_id="ACOUSTIC-REPEAT",
+        )
+        session.add(observation)
+        session.commit()
+        observation_id = observation.id
+
+        try:
+            backfill_acoustic_maturity.run(session)
+            backfill_acoustic_maturity.run(session)
+
+            session.refresh(observation)
+            assert observation.data_maturity == backfill_acoustic_maturity.MATURITY
+        finally:
+            session.execute(
+                delete(TransducerObservation).where(
+                    TransducerObservation.id == observation_id
+                )
+            )
+            session.commit()