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