From 1374a6cb013d438ee75263ee96067850185ff56f Mon Sep 17 00:00:00 2001 From: "jake.ross" Date: Wed, 7 Jan 2026 22:18:39 -0700 Subject: [PATCH 1/2] feat: add acoustic legacy fields to transducer observations --- admin/views/transducer_observation.py | 70 +++++++ ...b9770721_add_acoustic_legacy_fields_to_.py | 195 ++++++++++++++++++ db/transducer.py | 42 ++++ transfers/waterlevels_transducer_transfer.py | 26 +++ 4 files changed, 333 insertions(+) create mode 100644 alembic/versions/8ed4b9770721_add_acoustic_legacy_fields_to_.py diff --git a/admin/views/transducer_observation.py b/admin/views/transducer_observation.py index f0aeca787..351d970f0 100644 --- a/admin/views/transducer_observation.py +++ b/admin/views/transducer_observation.py @@ -88,6 +88,20 @@ class TransducerObservationAdmin(OcotilloModelView): "nma_waterlevelscontinuous_pressure_updated", "nma_waterlevelscontinuous_pressure_water_head", "nma_waterlevelscontinuous_pressure_water_head_adjusted", + "nma_waterlevelscontinuous_acoustic_created", + "nma_waterlevelscontinuous_acoustic_data_source", + "nma_waterlevelscontinuous_acoustic_global_id", + "nma_waterlevelscontinuous_acoustic_measurement_method", + "nma_waterlevelscontinuous_acoustic_measuring_agency", + "nma_waterlevelscontinuous_acoustic_notes", + "nma_waterlevelscontinuous_acoustic_point_id", + "nma_waterlevelscontinuous_acoustic_pre_process_data_field", + "nma_waterlevelscontinuous_acoustic_public_release", + "nma_waterlevelscontinuous_acoustic_sensor_hgt_above_mp", + "nma_waterlevelscontinuous_acoustic_serial_no", + "nma_waterlevelscontinuous_acoustic_server_receipt_date", + "nma_waterlevelscontinuous_acoustic_speaker_to_mic_length", + "nma_waterlevelscontinuous_acoustic_temperature_air", "created_at", "created_by_id", "created_by_name", @@ -116,6 +130,20 @@ class TransducerObservationAdmin(OcotilloModelView): "nma_waterlevelscontinuous_pressure_updated", "nma_waterlevelscontinuous_pressure_water_head", "nma_waterlevelscontinuous_pressure_water_head_adjusted", + "nma_waterlevelscontinuous_acoustic_created", + "nma_waterlevelscontinuous_acoustic_data_source", + "nma_waterlevelscontinuous_acoustic_global_id", + "nma_waterlevelscontinuous_acoustic_measurement_method", + "nma_waterlevelscontinuous_acoustic_measuring_agency", + "nma_waterlevelscontinuous_acoustic_notes", + "nma_waterlevelscontinuous_acoustic_point_id", + "nma_waterlevelscontinuous_acoustic_pre_process_data_field", + "nma_waterlevelscontinuous_acoustic_public_release", + "nma_waterlevelscontinuous_acoustic_sensor_hgt_above_mp", + "nma_waterlevelscontinuous_acoustic_serial_no", + "nma_waterlevelscontinuous_acoustic_server_receipt_date", + "nma_waterlevelscontinuous_acoustic_speaker_to_mic_length", + "nma_waterlevelscontinuous_acoustic_temperature_air", ] exclude_fields_from_edit = [ @@ -137,6 +165,20 @@ class TransducerObservationAdmin(OcotilloModelView): "nma_waterlevelscontinuous_pressure_updated", "nma_waterlevelscontinuous_pressure_water_head", "nma_waterlevelscontinuous_pressure_water_head_adjusted", + "nma_waterlevelscontinuous_acoustic_created", + "nma_waterlevelscontinuous_acoustic_data_source", + "nma_waterlevelscontinuous_acoustic_global_id", + "nma_waterlevelscontinuous_acoustic_measurement_method", + "nma_waterlevelscontinuous_acoustic_measuring_agency", + "nma_waterlevelscontinuous_acoustic_notes", + "nma_waterlevelscontinuous_acoustic_point_id", + "nma_waterlevelscontinuous_acoustic_pre_process_data_field", + "nma_waterlevelscontinuous_acoustic_public_release", + "nma_waterlevelscontinuous_acoustic_sensor_hgt_above_mp", + "nma_waterlevelscontinuous_acoustic_serial_no", + "nma_waterlevelscontinuous_acoustic_server_receipt_date", + "nma_waterlevelscontinuous_acoustic_speaker_to_mic_length", + "nma_waterlevelscontinuous_acoustic_temperature_air", ] readonly_fields = [ @@ -154,6 +196,20 @@ class TransducerObservationAdmin(OcotilloModelView): "nma_waterlevelscontinuous_pressure_updated", "nma_waterlevelscontinuous_pressure_water_head", "nma_waterlevelscontinuous_pressure_water_head_adjusted", + "nma_waterlevelscontinuous_acoustic_created", + "nma_waterlevelscontinuous_acoustic_data_source", + "nma_waterlevelscontinuous_acoustic_global_id", + "nma_waterlevelscontinuous_acoustic_measurement_method", + "nma_waterlevelscontinuous_acoustic_measuring_agency", + "nma_waterlevelscontinuous_acoustic_notes", + "nma_waterlevelscontinuous_acoustic_point_id", + "nma_waterlevelscontinuous_acoustic_pre_process_data_field", + "nma_waterlevelscontinuous_acoustic_public_release", + "nma_waterlevelscontinuous_acoustic_sensor_hgt_above_mp", + "nma_waterlevelscontinuous_acoustic_serial_no", + "nma_waterlevelscontinuous_acoustic_server_receipt_date", + "nma_waterlevelscontinuous_acoustic_speaker_to_mic_length", + "nma_waterlevelscontinuous_acoustic_temperature_air", ] labels = { @@ -177,6 +233,20 @@ class TransducerObservationAdmin(OcotilloModelView): "nma_waterlevelscontinuous_pressure_updated": "Updated", "nma_waterlevelscontinuous_pressure_water_head": "Water Head", "nma_waterlevelscontinuous_pressure_water_head_adjusted": "Water Head Adjusted", + "nma_waterlevelscontinuous_acoustic_created": "Acoustic Created", + "nma_waterlevelscontinuous_acoustic_data_source": "Acoustic Data Source", + "nma_waterlevelscontinuous_acoustic_global_id": "Acoustic Global ID", + "nma_waterlevelscontinuous_acoustic_measurement_method": "Acoustic Measurement Method", + "nma_waterlevelscontinuous_acoustic_measuring_agency": "Acoustic Measuring Agency", + "nma_waterlevelscontinuous_acoustic_notes": "Acoustic Notes", + "nma_waterlevelscontinuous_acoustic_point_id": "Acoustic Point ID", + "nma_waterlevelscontinuous_acoustic_pre_process_data_field": "Acoustic Pre-Process Data Field", + "nma_waterlevelscontinuous_acoustic_public_release": "Acoustic Public Release", + "nma_waterlevelscontinuous_acoustic_sensor_hgt_above_mp": "Acoustic Sensor Height Above MP", + "nma_waterlevelscontinuous_acoustic_serial_no": "Acoustic Serial No", + "nma_waterlevelscontinuous_acoustic_server_receipt_date": "Acoustic Server Receipt Date", + "nma_waterlevelscontinuous_acoustic_speaker_to_mic_length": "Acoustic Speaker To Mic Length", + "nma_waterlevelscontinuous_acoustic_temperature_air": "Acoustic Temperature Air", "created_at": "Created At", "created_by_name": "Created By", "updated_by_name": "Updated By", diff --git a/alembic/versions/8ed4b9770721_add_acoustic_legacy_fields_to_.py b/alembic/versions/8ed4b9770721_add_acoustic_legacy_fields_to_.py new file mode 100644 index 000000000..97a360fdb --- /dev/null +++ b/alembic/versions/8ed4b9770721_add_acoustic_legacy_fields_to_.py @@ -0,0 +1,195 @@ +"""add acoustic legacy fields to transducer observations + +Revision ID: 8ed4b9770721 +Revises: 1680a4a7cb77 +Create Date: 2026-01-07 22:12:20.045062 + +""" + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = "8ed4b9770721" +down_revision: Union[str, Sequence[str], None] = "1680a4a7cb77" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema.""" + op.add_column( + "transducer_observation", + sa.Column( + "nma_waterlevelscontinuous_acoustic_created", + sa.DateTime(timezone=True), + nullable=True, + ), + ) + op.add_column( + "transducer_observation", + sa.Column( + "nma_waterlevelscontinuous_acoustic_data_source", + sa.String(length=5), + nullable=True, + ), + ) + op.add_column( + "transducer_observation", + sa.Column( + "nma_waterlevelscontinuous_acoustic_global_id", + sa.String(length=40), + nullable=True, + ), + ) + op.add_column( + "transducer_observation", + sa.Column( + "nma_waterlevelscontinuous_acoustic_measurement_method", + sa.String(length=2), + nullable=True, + ), + ) + op.add_column( + "transducer_observation", + sa.Column( + "nma_waterlevelscontinuous_acoustic_measuring_agency", + sa.String(length=50), + nullable=True, + ), + ) + op.add_column( + "transducer_observation", + sa.Column( + "nma_waterlevelscontinuous_acoustic_notes", + sa.String(length=200), + nullable=True, + ), + ) + op.add_column( + "transducer_observation", + sa.Column( + "nma_waterlevelscontinuous_acoustic_point_id", + sa.String(length=50), + nullable=True, + ), + ) + op.add_column( + "transducer_observation", + sa.Column( + "nma_waterlevelscontinuous_acoustic_pre_process_data_field", + sa.Float(), + nullable=True, + ), + ) + op.add_column( + "transducer_observation", + sa.Column( + "nma_waterlevelscontinuous_acoustic_public_release", + sa.Boolean(), + nullable=True, + ), + ) + op.add_column( + "transducer_observation", + sa.Column( + "nma_waterlevelscontinuous_acoustic_sensor_hgt_above_mp", + sa.Float(), + nullable=True, + ), + ) + op.add_column( + "transducer_observation", + sa.Column( + "nma_waterlevelscontinuous_acoustic_serial_no", + sa.String(length=50), + nullable=True, + ), + ) + op.add_column( + "transducer_observation", + sa.Column( + "nma_waterlevelscontinuous_acoustic_server_receipt_date", + sa.DateTime(timezone=True), + nullable=True, + ), + ) + op.add_column( + "transducer_observation", + sa.Column( + "nma_waterlevelscontinuous_acoustic_speaker_to_mic_length", + sa.Float(), + nullable=True, + ), + ) + op.add_column( + "transducer_observation", + sa.Column( + "nma_waterlevelscontinuous_acoustic_temperature_air", + sa.Float(), + nullable=True, + ), + ) + + +def downgrade() -> None: + """Downgrade schema.""" + op.drop_column( + "transducer_observation", + "nma_waterlevelscontinuous_acoustic_temperature_air", + ) + op.drop_column( + "transducer_observation", + "nma_waterlevelscontinuous_acoustic_speaker_to_mic_length", + ) + op.drop_column( + "transducer_observation", + "nma_waterlevelscontinuous_acoustic_server_receipt_date", + ) + op.drop_column( + "transducer_observation", + "nma_waterlevelscontinuous_acoustic_serial_no", + ) + op.drop_column( + "transducer_observation", + "nma_waterlevelscontinuous_acoustic_sensor_hgt_above_mp", + ) + op.drop_column( + "transducer_observation", + "nma_waterlevelscontinuous_acoustic_public_release", + ) + op.drop_column( + "transducer_observation", + "nma_waterlevelscontinuous_acoustic_pre_process_data_field", + ) + op.drop_column( + "transducer_observation", + "nma_waterlevelscontinuous_acoustic_point_id", + ) + op.drop_column( + "transducer_observation", + "nma_waterlevelscontinuous_acoustic_notes", + ) + op.drop_column( + "transducer_observation", + "nma_waterlevelscontinuous_acoustic_measuring_agency", + ) + op.drop_column( + "transducer_observation", + "nma_waterlevelscontinuous_acoustic_measurement_method", + ) + op.drop_column( + "transducer_observation", + "nma_waterlevelscontinuous_acoustic_global_id", + ) + op.drop_column( + "transducer_observation", + "nma_waterlevelscontinuous_acoustic_data_source", + ) + op.drop_column( + "transducer_observation", + "nma_waterlevelscontinuous_acoustic_created", + ) diff --git a/db/transducer.py b/db/transducer.py index 608260ce5..ae9ac01da 100644 --- a/db/transducer.py +++ b/db/transducer.py @@ -162,6 +162,48 @@ class TransducerObservation(Base, AutoBaseMixin, ReleaseMixin): nma_waterlevelscontinuous_pressure_water_head_adjusted: Mapped[float] = ( mapped_column(Float, nullable=True) ) + nma_waterlevelscontinuous_acoustic_created: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=True + ) + nma_waterlevelscontinuous_acoustic_data_source: Mapped[str] = mapped_column( + String(5), nullable=True + ) + nma_waterlevelscontinuous_acoustic_global_id: Mapped[str] = mapped_column( + String(40), nullable=True + ) + nma_waterlevelscontinuous_acoustic_measurement_method: Mapped[str] = mapped_column( + String(2), nullable=True + ) + nma_waterlevelscontinuous_acoustic_measuring_agency: Mapped[str] = mapped_column( + String(50), nullable=True + ) + nma_waterlevelscontinuous_acoustic_notes: Mapped[str] = mapped_column( + String(200), nullable=True + ) + nma_waterlevelscontinuous_acoustic_point_id: Mapped[str] = mapped_column( + String(50), nullable=True + ) + nma_waterlevelscontinuous_acoustic_pre_process_data_field: Mapped[float] = ( + mapped_column(Float, nullable=True) + ) + nma_waterlevelscontinuous_acoustic_public_release: Mapped[bool] = mapped_column( + Boolean, nullable=True + ) + nma_waterlevelscontinuous_acoustic_sensor_hgt_above_mp: Mapped[float] = ( + mapped_column(Float, nullable=True) + ) + nma_waterlevelscontinuous_acoustic_serial_no: Mapped[str] = mapped_column( + String(50), nullable=True + ) + nma_waterlevelscontinuous_acoustic_server_receipt_date: Mapped[datetime] = ( + mapped_column(DateTime(timezone=True), nullable=True) + ) + nma_waterlevelscontinuous_acoustic_speaker_to_mic_length: Mapped[float] = ( + mapped_column(Float, nullable=True) + ) + nma_waterlevelscontinuous_acoustic_temperature_air: Mapped[float] = mapped_column( + Float, nullable=True + ) # qc_block_id: Mapped[Optional[int]] = mapped_column( # ForeignKey("transducer_observation_block.id", ondelete="SET NULL"), index=True diff --git a/transfers/waterlevels_transducer_transfer.py b/transfers/waterlevels_transducer_transfer.py index 11048c793..b2427d1b6 100644 --- a/transfers/waterlevels_transducer_transfer.py +++ b/transfers/waterlevels_transducer_transfer.py @@ -233,6 +233,32 @@ def val(key: str): "nma_waterlevelscontinuous_pressure_water_head_adjusted": val( "WaterHeadAdjusted" ), + "nma_waterlevelscontinuous_acoustic_created": val("Created"), + "nma_waterlevelscontinuous_acoustic_data_source": val("DataSource"), + "nma_waterlevelscontinuous_acoustic_global_id": val("GlobalID"), + "nma_waterlevelscontinuous_acoustic_measurement_method": val( + "MeasurementMethod" + ), + "nma_waterlevelscontinuous_acoustic_measuring_agency": val( + "MeasuringAgency" + ), + "nma_waterlevelscontinuous_acoustic_notes": val("Notes"), + "nma_waterlevelscontinuous_acoustic_point_id": val("PointID"), + "nma_waterlevelscontinuous_acoustic_pre_process_data_field": val( + "PreProcessDataField" + ), + "nma_waterlevelscontinuous_acoustic_public_release": val("PublicRelease"), + "nma_waterlevelscontinuous_acoustic_sensor_hgt_above_mp": val( + "SensorHgtAboveMP" + ), + "nma_waterlevelscontinuous_acoustic_serial_no": val("SerialNo"), + "nma_waterlevelscontinuous_acoustic_server_receipt_date": val( + "ServerReceiptDate" + ), + "nma_waterlevelscontinuous_acoustic_speaker_to_mic_length": val( + "SpeakerToMicLength" + ), + "nma_waterlevelscontinuous_acoustic_temperature_air": val("TemperatureAir"), } @staticmethod From 2d844191b6e685dc1a179088bd1bf8390bd24543 Mon Sep 17 00:00:00 2001 From: "jake.ross" Date: Wed, 7 Jan 2026 22:29:31 -0700 Subject: [PATCH 2/2] feat: refactor legacy payload methods for water levels data handling --- transfers/waterlevels_transducer_transfer.py | 132 +++++++++++-------- 1 file changed, 74 insertions(+), 58 deletions(-) diff --git a/transfers/waterlevels_transducer_transfer.py b/transfers/waterlevels_transducer_transfer.py index b2427d1b6..3deebc047 100644 --- a/transfers/waterlevels_transducer_transfer.py +++ b/transfers/waterlevels_transducer_transfer.py @@ -13,6 +13,7 @@ # See the License for the specific language governing permissions and # limitations under the License. # =============================================================================== +from typing import Any import pandas as pd from pandas import Timestamp @@ -201,65 +202,16 @@ def _make_observation( self._capture_error(pointid, str(e), "DepthToWaterBGS") def _legacy_payload(self, row: pd.Series) -> dict: - def val(key: str): - if key not in self._df_columns: - return None - field = self._itertuples_field_map.get(key, key) - v = getattr(row, field, None) - if pd.isna(v): - return None - return v + return {} - return { - "nma_waterlevelscontinuous_pressure_conddl_ms_cm": val("CONDDL (mS/cm)"), - "nma_waterlevelscontinuous_pressure_checked_by": val("CheckedBy"), - "nma_waterlevelscontinuous_pressure_created": val("Created"), - "nma_waterlevelscontinuous_pressure_data_source": val("DataSource"), - "nma_waterlevelscontinuous_pressure_global_id": val("GlobalID"), - "nma_waterlevelscontinuous_pressure_measurement_method": val( - "MeasurementMethod" - ), - "nma_waterlevelscontinuous_pressure_measuring_agency": val( - "MeasuringAgency" - ), - "nma_waterlevelscontinuous_pressure_notes": val("Notes"), - "nma_waterlevelscontinuous_pressure_processed_by": val("ProcessedBy"), - "nma_waterlevelscontinuous_pressure_qced": val("QCed"), - "nma_waterlevelscontinuous_pressure_temperature_water": val( - "TemperatureWater" - ), - "nma_waterlevelscontinuous_pressure_updated": val("Updated"), - "nma_waterlevelscontinuous_pressure_water_head": val("WaterHead"), - "nma_waterlevelscontinuous_pressure_water_head_adjusted": val( - "WaterHeadAdjusted" - ), - "nma_waterlevelscontinuous_acoustic_created": val("Created"), - "nma_waterlevelscontinuous_acoustic_data_source": val("DataSource"), - "nma_waterlevelscontinuous_acoustic_global_id": val("GlobalID"), - "nma_waterlevelscontinuous_acoustic_measurement_method": val( - "MeasurementMethod" - ), - "nma_waterlevelscontinuous_acoustic_measuring_agency": val( - "MeasuringAgency" - ), - "nma_waterlevelscontinuous_acoustic_notes": val("Notes"), - "nma_waterlevelscontinuous_acoustic_point_id": val("PointID"), - "nma_waterlevelscontinuous_acoustic_pre_process_data_field": val( - "PreProcessDataField" - ), - "nma_waterlevelscontinuous_acoustic_public_release": val("PublicRelease"), - "nma_waterlevelscontinuous_acoustic_sensor_hgt_above_mp": val( - "SensorHgtAboveMP" - ), - "nma_waterlevelscontinuous_acoustic_serial_no": val("SerialNo"), - "nma_waterlevelscontinuous_acoustic_server_receipt_date": val( - "ServerReceiptDate" - ), - "nma_waterlevelscontinuous_acoustic_speaker_to_mic_length": val( - "SpeakerToMicLength" - ), - "nma_waterlevelscontinuous_acoustic_temperature_air": val("TemperatureAir"), - } + def _legacy_val(self, row: pd.Series, key: str) -> Any: + if key not in self._df_columns: + return None + field = self._itertuples_field_map.get(key, key) + v = getattr(row, field, None) + if pd.isna(v): + return None + return v @staticmethod def _build_itertuples_field_map(df: pd.DataFrame) -> dict[str, str]: @@ -285,12 +237,76 @@ class WaterLevelsContinuousPressureTransferer(WaterLevelsContinuousTransferer): _partition_field = "QCed" _sensor_types = ("Pressure Transducer", "Barometer", "DiverLink", "Diver Cable") + def _legacy_payload(self, row: pd.Series) -> dict: + val = self._legacy_val + return { + "nma_waterlevelscontinuous_pressure_conddl_ms_cm": val( + row, "CONDDL (mS/cm)" + ), + "nma_waterlevelscontinuous_pressure_checked_by": val(row, "CheckedBy"), + "nma_waterlevelscontinuous_pressure_created": val(row, "Created"), + "nma_waterlevelscontinuous_pressure_data_source": val(row, "DataSource"), + "nma_waterlevelscontinuous_pressure_global_id": val(row, "GlobalID"), + "nma_waterlevelscontinuous_pressure_measurement_method": val( + row, "MeasurementMethod" + ), + "nma_waterlevelscontinuous_pressure_measuring_agency": val( + row, "MeasuringAgency" + ), + "nma_waterlevelscontinuous_pressure_notes": val(row, "Notes"), + "nma_waterlevelscontinuous_pressure_processed_by": val(row, "ProcessedBy"), + "nma_waterlevelscontinuous_pressure_qced": val(row, "QCed"), + "nma_waterlevelscontinuous_pressure_temperature_water": val( + row, "TemperatureWater" + ), + "nma_waterlevelscontinuous_pressure_updated": val(row, "Updated"), + "nma_waterlevelscontinuous_pressure_water_head": val(row, "WaterHead"), + "nma_waterlevelscontinuous_pressure_water_head_adjusted": val( + row, "WaterHeadAdjusted" + ), + } + class WaterLevelsContinuousAcousticTransferer(WaterLevelsContinuousTransferer): source_table = "WaterLevelsContinuous_Acoustic" _partition_field = "PublicRelease" _sensor_types = ("Acoustic Sounder",) + def _legacy_payload(self, row: pd.Series) -> dict: + val = self._legacy_val + return { + "nma_waterlevelscontinuous_acoustic_created": val(row, "Created"), + "nma_waterlevelscontinuous_acoustic_data_source": val(row, "DataSource"), + "nma_waterlevelscontinuous_acoustic_global_id": val(row, "GlobalID"), + "nma_waterlevelscontinuous_acoustic_measurement_method": val( + row, "MeasurementMethod" + ), + "nma_waterlevelscontinuous_acoustic_measuring_agency": val( + row, "MeasuringAgency" + ), + "nma_waterlevelscontinuous_acoustic_notes": val(row, "Notes"), + "nma_waterlevelscontinuous_acoustic_point_id": val(row, "PointID"), + "nma_waterlevelscontinuous_acoustic_pre_process_data_field": val( + row, "PreProcessDataField" + ), + "nma_waterlevelscontinuous_acoustic_public_release": val( + row, "PublicRelease" + ), + "nma_waterlevelscontinuous_acoustic_sensor_hgt_above_mp": val( + row, "SensorHgtAboveMP" + ), + "nma_waterlevelscontinuous_acoustic_serial_no": val(row, "SerialNo"), + "nma_waterlevelscontinuous_acoustic_server_receipt_date": val( + row, "ServerReceiptDate" + ), + "nma_waterlevelscontinuous_acoustic_speaker_to_mic_length": val( + row, "SpeakerToMicLength" + ), + "nma_waterlevelscontinuous_acoustic_temperature_air": val( + row, "TemperatureAir" + ), + } + def _find_deployment(ts, deployments): date = ts.date()