Merge pull request #519 from DataIntegrationGroup/transfer-fix

transfer-fix

Author: jirhiker

transfers/waterlevels_transfer.py

@@ -41,6 +41,7 @@
     filter_by_valid_measuring_agency,
     lexicon_mapper,
     get_transfers_data_path,
+    replace_nans,
 )
 
 # constants
@@ -94,6 +95,7 @@ def __init__(self, *args, **kw):
 
     def _get_dfs(self) -> tuple[pd.DataFrame, pd.DataFrame]:
         input_df = read_csv(self.source_table, dtype={"MeasuredBy": str})
+        input_df = replace_nans(input_df)
         cleaned_df = filter_to_valid_point_ids(input_df)
         cleaned_df = filter_by_valid_measuring_agency(cleaned_df)
         logger.info(
@@ -314,11 +316,48 @@ def _make_observation(
         if dq_raw and pd.notna(dq_raw):
             dq_code = str(dq_raw).strip()
             try:
-                data_quality = lexicon_mapper.map_value(f"LU_DataQuality:{dq_code}")
+                mapped_quality = lexicon_mapper.map_value(f"LU_DataQuality:{dq_code}")
+                if pd.isna(mapped_quality):
+                    logger.warning(
+                        "%sMapped DataQuality '%s' to NaN for WaterLevels record %s; "
+                        "storing NULL to satisfy FK",
+                        SPACE_6,
+                        dq_code,
+                        row.GlobalID,
+                    )
+                    self._capture_error(
+                        row.PointID,
+                        f"Mapped DataQuality '{dq_code}' to NaN; stored NULL",
+                        "DataQuality",
+                    )
+                    data_quality = None
+                else:
+                    mapped_quality_text = str(mapped_quality).strip()
+                    if mapped_quality_text and mapped_quality_text.lower() != "nan":
+                        data_quality = mapped_quality_text
+                    else:
+                        logger.warning(
+                            "%sMapped DataQuality '%s' to empty value for WaterLevels "
+                            "record %s; storing NULL to satisfy FK",
+                            SPACE_6,
+                            dq_code,
+                            row.GlobalID,
+                        )
+                        self._capture_error(
+                            row.PointID,
+                            f"Mapped DataQuality '{dq_code}' to empty value; stored NULL",
+                            "DataQuality",
+                        )
+                        data_quality = None
             except KeyError:
                 logger.warning(
                     f"{SPACE_6}Unknown DataQuality code '{dq_code}' for WaterLevels record {row.GlobalID}"
                 )
+                self._capture_error(
+                    row.PointID,
+                    f"Unknown DataQuality code '{dq_code}'",
+                    "DataQuality",
+                )
 
             # TODO: after sensors have been added to the database update sensor_id (or sensor) for waterlevels that come from db sensors (like e probes?)
         observation = Observation(