fix: add new well status term and normalize completion date handling

Author: jirhiker

core/lexicon.json

@@ -2260,7 +2260,12 @@
       "categories": ["status_value"],
       "term": "Open",
       "definition": "The well is open."
-    },  
+    },
+    {
+        "categories": ["status_value"],
+        "term": "Open (unequipped)",
+        "definition": "The well is open and unequipped."
+    },
     {
       "categories": ["status_value"],
       "term": "Closed",

tests/test_transfer_legacy_dates.py

@@ -28,6 +28,7 @@
 import pytest
 
 from db import Sample
+from transfers.well_transfer import _normalize_completion_date
 from transfers.util import make_location
 from transfers.waterlevels_transfer import WaterLevelTransferer
 
@@ -207,6 +208,18 @@ def test_make_observation_maps_data_quality():
         assert observation.nma_data_quality == "Mapped Quality"
 
 
+def test_normalize_completion_date_drops_time_from_datetime():
+    value = datetime.datetime(2024, 7, 3, 14, 15, 16)
+    assert _normalize_completion_date(value) == datetime.date(2024, 7, 3)
+
+
+def test_normalize_completion_date_drops_time_from_timestamp_and_string():
+    ts_value = pd.Timestamp("2021-05-06 23:59:00")
+    str_value = "2021-05-06 23:59:00.000"
+    assert _normalize_completion_date(ts_value) == datetime.date(2021, 5, 6)
+    assert _normalize_completion_date(str_value) == datetime.date(2021, 5, 6)
+
+
 def test_get_dt_utc_respects_time_datum():
     transfer = WaterLevelTransferer.__new__(WaterLevelTransferer)
     transfer.errors = []

transfers/util.py

@@ -57,6 +57,38 @@
 }
 
 
+DEFINED_RECORDING_INTERVALS = {
+    "SA-0174": (1, "hour"),
+    "SO-0140": (0.25, "hour"),
+    "SO-0145": (0.25, "hour"),
+    "SO-0146": (0.25, "hour"),
+    "SO-0148": (0.25, "hour"),
+    "SO-0160": (0.25, "hour"),
+    "SO-0163": (0.25, "hour"),
+    "SO-0165": (0.25, "hour"),
+    "SO-0166": (0.25, "hour"),
+    "SO-0175": (0.25, "hour"),
+    "SO-0177": (0.25, "hour"),
+    "SO-0189": (0.25, "hour"),
+    "SO-0191": (0.25, "hour"),
+    "SO-0194": (0.25, "hour"),
+    "SO-0200": (0.25, "hour"),
+    "SO-0204": (0.25, "hour"),
+    "SO-0224": (0.25, "hour"),
+    "SO-0238": (0.25, "hour"),
+    "SO-0247": (0.25, "hour"),
+    "SO-0249": (0.25, "hour"),
+    "SO-0261": (0.25, "hour"),
+    "SM-0055": (6.0, "hour"),
+    "SM-0259": (12, "hour"),
+    "HS-038": (12, "hour"),
+    "EB-220": (12, "hour"),
+    "SO-0144": (0.25, "hour"),
+    "SO-0142": (0.25, "hour"),
+    "SO-0190": (0.25, "hour"),
+}
+
+
 class MeasuringPointEstimator:
     def __init__(self):
         df = read_csv("WaterLevels")
@@ -123,6 +155,12 @@ def estimate_measuring_point_height(
         return mphs, mph_descs, start_dates, end_dates
 
 
+def _get_defined_recording_interval(pointid: str) -> tuple[int, str] | None:
+    if pointid in DEFINED_RECORDING_INTERVALS:
+        return DEFINED_RECORDING_INTERVALS[pointid]
+    return None
+
+
 class SensorParameterEstimator:
     def __init__(self, sensor_type: str):
         if sensor_type == "Pressure Transducer":
@@ -156,7 +194,16 @@ def estimate_recording_interval(
         installation_date: datetime = None,
         removal_date: datetime = None,
     ) -> tuple[int | None, str | None, str | None]:
+        """
+        return estimated recording interval, unit, and error message if applicable
+        """
         point_id = record.PointID
+
+        # get statically defined recording interval provided by Ethan
+        ri = _get_defined_recording_interval(point_id)
+        if ri is not None:
+            return ri[0], ri[1], None
+
         cdf = self._get_values(point_id)
         if len(cdf) == 0:
             return None, None, f"No measurements found for PointID: {point_id}"

transfers/well_transfer.py

@@ -18,7 +18,7 @@
 import time
 import traceback
 from concurrent.futures import ThreadPoolExecutor, as_completed
-from datetime import datetime, UTC
+from datetime import date, datetime, UTC
 from zoneinfo import ZoneInfo
 
 import pandas as pd
@@ -95,6 +95,27 @@
 ]
 
 
+def _normalize_completion_date(value):
+    if value is None or pd.isna(value):
+        return None
+
+    if isinstance(value, pd.Timestamp):
+        return value.date()
+
+    if isinstance(value, datetime):
+        return value.date()
+
+    if isinstance(value, date):
+        return value
+
+    if isinstance(value, str):
+        parsed = pd.to_datetime(value.strip(), errors="coerce")
+        if not pd.isna(parsed):
+            return parsed.date()
+
+    return value
+
+
 class WellTransferer(Transferer):
     source_table = "WellData"
 
@@ -279,123 +300,6 @@ def _get_dfs(self):
             cleaned_df = cleaned_df[cleaned_df["PointID"].isin(self.pointids)]
         return input_df, cleaned_df
 
-    # def _step(self, session: Session, df: pd.DataFrame, i: int, row: pd.Series):
-    #
-    #     try:
-    #         first_visit_date = get_first_visit_date(row)
-    #         well_purposes = (
-    #             [] if isna(row.CurrentUse) else self._extract_well_purposes(row)
-    #         )
-    #         well_casing_materials = (
-    #             [] if isna(row.CasingDescription) else extract_casing_materials(row)
-    #         )
-    #         well_pump_type = extract_well_pump_type(row)
-    #
-    #         wcm = None
-    #         if notna(row.ConstructionMethod):
-    #             wcm = self._get_lexicon_value(
-    #                 row, f"LU_ConstructionMethod:{row.ConstructionMethod}", "Unknown"
-    #             )
-    #
-    #         mpheight = row.MPHeight
-    #         mpheight_description = row.MeasuringPoint
-    #         if mpheight is None:
-    #             mphs = self._measuring_point_estimator.estimate_measuring_point_height(
-    #                 row
-    #             )
-    #             if mphs:
-    #                 try:
-    #                     mpheight = mphs[0][0]
-    #                     mpheight_description = mphs[1][0]
-    #                 except IndexError:
-    #                     if self.verbose:
-    #                         logger.warning(
-    #                             f"Measuring point height estimation failed for well {row.PointID}, {mphs}"
-    #                         )
-    #
-    #         data = CreateWell(
-    #             location_id=0,
-    #             name=row.PointID,
-    #             first_visit_date=first_visit_date,
-    #             hole_depth=row.HoleDepth,
-    #             well_depth=row.WellDepth,
-    #             well_casing_diameter=(
-    #                 row.CasingDiameter * 12 if row.CasingDiameter else None
-    #             ),
-    #             well_casing_depth=row.CasingDepth,
-    #             release_status="public" if row.PublicRelease else "private",
-    #             measuring_point_height=mpheight,
-    #             measuring_point_description=mpheight_description,
-    #             notes=(
-    #                 [{"content": row.Notes, "note_type": "General"}]
-    #                 if row.Notes
-    #                 else []
-    #             ),
-    #             well_completion_date=row.CompletionDate,
-    #             well_driller_name=row.DrillerName,
-    #             well_construction_method=wcm,
-    #             well_pump_type=well_pump_type,
-    #         )
-    #
-    #         CreateWell.model_validate(data)
-    #     except ValidationError as e:
-    #         self._capture_validation_error(row.PointID, e)
-    #         return
-    #
-    #     well = None
-    #     try:
-    #         well_data = data.model_dump(exclude=EXCLUDED_FIELDS)
-    #         well_data["thing_type"] = "water well"
-    #         well_data["nma_pk_welldata"] = row.WellID
-    #         well_data["nma_pk_location"] = row.LocationId
-    #
-    #         well = Thing(**well_data)
-    #         session.add(well)
-    #
-    #         if well_purposes:
-    #             for wp in well_purposes:
-    #                 # TODO: add validation logic here
-    #                 if wp in WellPurposeEnum:
-    #                     wp_obj = WellPurpose(thing=well, purpose=wp)
-    #                     session.add(wp_obj)
-    #                 else:
-    #                     logger.critical(f"{well.name}. Invalid well purpose: {wp}")
-    #
-    #         if well_casing_materials:
-    #             for wcm in well_casing_materials:
-    #                 # TODO: add validation logic here
-    #                 if wcm in WellCasingMaterialEnum:
-    #                     wcm_obj = WellCasingMaterial(thing=well, material=wcm)
-    #                     session.add(wcm_obj)
-    #                 else:
-    #                     logger.critical(
-    #                         f"{well.name}. Invalid well casing material: {wcm}"
-    #                     )
-    #     except Exception as e:
-    #         if well is not None:
-    #             session.expunge(well)
-    #
-    #         self._capture_error(row.PointID, str(e), "UnknownField")
-    #
-    #         logger.critical(f"Error creating well for {row.PointID}: {e}")
-    #         return
-    #
-    #     try:
-    #         location, elevation_method, notes = make_location(
-    #             row, self._cached_elevations
-    #         )
-    #         session.add(location)
-    #         # session.flush()
-    #         self._added_locations[row.PointID] = (elevation_method, notes)
-    #     except Exception as e:
-    #         import traceback
-    #
-    #         traceback.print_exc()
-    #         self._capture_error(row.PointID, str(e), str(e), "Location")
-    #         logger.critical(f"Error making location for {row.PointID}: {e}")
-    #
-    #         return
-    #
     def _extract_well_purposes(self, row) -> list[str]:
         cu = row.CurrentUse
 
@@ -659,7 +563,7 @@ def _build_well_payload(self, row) -> CreateWell | None:
                     if row.Notes
                     else []
                 ),
-                well_completion_date=row.CompletionDate,
+                well_completion_date=_normalize_completion_date(row.CompletionDate),
                 well_driller_name=row.DrillerName,
                 well_construction_method=wcm,
                 well_pump_type=well_pump_type,