From 7a3915f41a7eaa3da7f6886e0a93d3caa7cbdfe0 Mon Sep 17 00:00:00 2001 From: jakeross Date: Thu, 20 Aug 2026 16:44:30 -0700 Subject: [PATCH] feat(transducer): backfill data_maturity on acoustic observations Revision b2c3d4e5f6a7 backfilled data_maturity from nma_waterlevelscontinuous_pressure_qced. Acoustic readings have no such flag -- AMPAPI's WaterLevelsContinuous_Acoustic table has no QCed column -- so all 394,086 of them were skipped, which is the entire Wellntel record (BDMS-1169). Rows are matched on nma_waterlevelscontinuous_acoustic_global_id rather than on a NULL pressure flag: the global id is written by the acoustic transferer on every row and never by the pressure one, so it identifies provenance instead of merely the absence of evidence. Only rows whose data_maturity is still NULL are touched, so re-running is 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. The value itself is a recorded decision, not a derivation; there is no QC field in the acoustic legacy schema to read. The transfer's review_status='approved' blocks are not evidence for it, since those come from PublicRelease, which describes visibility rather than review. Co-Authored-By: Claude Opus 5 --- ...20_0001_backfill_acoustic_data_maturity.py | 88 +++++++++++++++++ tests/test_data_migrations.py | 96 ++++++++++++++++++- 2 files changed, 183 insertions(+), 1 deletion(-) create mode 100644 data_migrations/migrations/20260820_0001_backfill_acoustic_data_maturity.py diff --git a/data_migrations/migrations/20260820_0001_backfill_acoustic_data_maturity.py b/data_migrations/migrations/20260820_0001_backfill_acoustic_data_maturity.py new file mode 100644 index 000000000..7996e4b74 --- /dev/null +++ b/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 ============================================= diff --git a/tests/test_data_migrations.py b/tests/test_data_migrations.py index 8c11177d0..bf349711e 100644 --- a/tests/test_data_migrations.py +++ b/tests/test_data_migrations.py @@ -14,8 +14,9 @@ # 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" @@ -23,10 +24,15 @@ 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()