Acquisition logging:

* repair and log instead of rejecting files with invalid missing codes
* introduce dtype inference code in acquisition to see if further refactors possible

Author: dshemetov

src/acquisition/covidcast/csv_importer.py

@@ -212,7 +212,7 @@ def validate_quantity(row, attr_quantity):
       return "Error"
 
   @staticmethod
-  def validate_missing_code(row, attr_quantity, attr_name):
+  def validate_missing_code(row, attr_quantity, attr_name, filepath=None, logger=None):
     """Take a row and validate the missing code associated with
     a quantity (e.g., val, se, stderr).
 
@@ -221,27 +221,32 @@ def validate_missing_code(row, attr_quantity, attr_name):
     to infer missing codes except for a very simple cases; the default
     is to produce an error so that the issue can be fixed in indicators.
     """
-    if hasattr(row, "missing_" + attr_name):
-      missing_entry = getattr(row, "missing_" + attr_name)
+    if logger is None:
+      logger = get_structured_logger('load_csv')
+    missing_entry = getattr(row, "missing_" + attr_name, None)
+
+    if missing_entry is not None:
       try:
         missing_entry = int(float(missing_entry)) # convert from string to float to int
       except ValueError:
-        return None
-      # A missing code should never contradict the quantity being present,
-      # since that will be filtered in the export_to_csv util in
-      # covidcast-indicators; nonetheless this code is here for safety.
-      if attr_quantity is not None and missing_entry != Nans.NOT_MISSING.value:
-        return None
-      elif attr_quantity is None and missing_entry == Nans.NOT_MISSING.value:
-        return None
-      return missing_entry
-    else:
-      if attr_quantity is None:
-        return Nans.OTHER.value
+        missing_entry = None
+
+    if missing_entry is None and attr_quantity is not None:
       return Nans.NOT_MISSING.value
+    if missing_entry is None and attr_quantity is None:
+      return Nans.OTHER.value
+
+    if missing_entry != Nans.NOT_MISSING.value and attr_quantity is not None:
+      logger.warning(event = f"missing_{attr_name} column contradicting {attr_name} presence.", detail = (str(row)), file = filepath)
+      return Nans.NOT_MISSING.value
+    if missing_entry == Nans.NOT_MISSING.value and attr_quantity is None:
+      logger.warning(event = f"missing_{attr_name} column contradicting {attr_name} presence.", detail = (str(row)), file = filepath)
+      return Nans.OTHER.value
+
+    return missing_entry
 
   @staticmethod
-  def extract_and_check_row(row, geo_type):
+  def extract_and_check_row(row, geo_type, filepath=None):
     """Extract and return `RowValues` from a CSV row, with sanity checks.
 
     Also returns the name of the field which failed sanity check, or None.
@@ -316,15 +321,9 @@ def extract_and_check_row(row, geo_type):
       return (None, 'sample_size')
 
     # Validate and write missingness codes
-    missing_value = CsvImporter.validate_missing_code(row, value, "val")
-    if missing_value is None:
-      return (None, 'missing_val')
-    missing_stderr = CsvImporter.validate_missing_code(row, stderr, "se")
-    if missing_stderr is None:
-      return (None, 'missing_se')
-    missing_sample_size = CsvImporter.validate_missing_code(row, sample_size, "sample_size")
-    if missing_sample_size is None:
-      return (None, 'missing_sample_size')
+    missing_value = CsvImporter.validate_missing_code(row, value, "val", filepath)
+    missing_stderr = CsvImporter.validate_missing_code(row, stderr, "se", filepath)
+    missing_sample_size = CsvImporter.validate_missing_code(row, sample_size, "sample_size", filepath)
 
     # return extracted and validated row values
     row_values = CsvImporter.RowValues(
@@ -353,7 +352,7 @@ def load_csv(filepath, geo_type, pandas=pandas):
       return
 
     for row in table.itertuples(index=False):
-      row_values, error = CsvImporter.extract_and_check_row(row, geo_type)
+      row_values, error = CsvImporter.extract_and_check_row(row, geo_type, filepath)
       if error:
         logger.warning(event = 'invalid value for row', detail=(str(row), error), file=filepath)
         yield None

tests/acquisition/covidcast/test_csv_importer.py

@@ -200,12 +200,6 @@ def make_row(
       (make_row(val='val'), 'val'),
       (make_row(se='se'), 'se'),
       (make_row(sample_size='sample_size'), 'sample_size'),
-      (make_row(missing_val='missing_val'), 'missing_val'),
-      (make_row(missing_se='missing_val'), 'missing_se'),
-      (make_row(missing_sample_size='missing_val'), 'missing_sample_size'),
-      (make_row(val='1.2', missing_val=str(float(Nans.OTHER))), 'missing_val'),
-      (make_row(se='1.2', missing_se=str(float(Nans.OTHER))), 'missing_se'),
-      (make_row(sample_size='1.2', missing_sample_size=str(float(Nans.OTHER))), 'missing_sample_size'),
     ]
 
     for ((geo_type, row), field) in failure_cases:
@@ -216,7 +210,8 @@ def make_row(
     success_cases = [
       (make_row(), CsvImporter.RowValues('vi', 1.23, 4.56, 100.5, Nans.NOT_MISSING, Nans.NOT_MISSING, Nans.NOT_MISSING)),
       (make_row(val=None, se=np.nan, sample_size='', missing_val=str(float(Nans.DELETED)), missing_se=str(float(Nans.DELETED)), missing_sample_size=str(float(Nans.DELETED))), CsvImporter.RowValues('vi', None, None, None, Nans.DELETED, Nans.DELETED, Nans.DELETED)),
-      (make_row(se='', sample_size='NA', missing_se=str(float(Nans.OTHER)), missing_sample_size=str(float(Nans.OTHER))), CsvImporter.RowValues('vi', 1.23, None, None, Nans.NOT_MISSING, Nans.OTHER, Nans.OTHER))
+      (make_row(se='', sample_size='NA', missing_se=str(float(Nans.OTHER)), missing_sample_size=str(float(Nans.OTHER))), CsvImporter.RowValues('vi', 1.23, None, None, Nans.NOT_MISSING, Nans.OTHER, Nans.OTHER)),
+      (make_row(sample_size=None, missing_val='missing_val', missing_se=str(float(Nans.OTHER)), missing_sample_size=str(float(Nans.NOT_MISSING))), CsvImporter.RowValues('vi', 1.23, 4.56, None, Nans.NOT_MISSING, Nans.NOT_MISSING, Nans.OTHER)),
     ]
 
     for ((geo_type, row), field) in success_cases:
@@ -281,9 +276,7 @@ def test_load_csv_with_valid_header(self):
 
     self.assertIsNone(rows[3])
 
-    # now with missing values! the last missing_sample_size
-    # contains an error code while data is available, which
-    # should give an error
+    # now with missing values!
     data = {
       'geo_id': ['ca', 'tx', 'fl', 'ak'],
       'val': [np.nan, '1.2', '1.3', '1.4'],
@@ -328,4 +321,10 @@ def test_load_csv_with_valid_header(self):
     self.assertEqual(rows[2].missing_stderr, Nans.NOT_MISSING)
     self.assertEqual(rows[2].missing_sample_size, Nans.REGION_EXCEPTION)
 
-    self.assertIsNone(rows[3])
+    self.assertEqual(rows[3].geo_value, 'ak')
+    self.assertEqual(rows[3].value, 1.4)
+    self.assertEqual(rows[3].stderr, 2.4)
+    self.assertEqual(rows[3].sample_size, 304)
+    self.assertEqual(rows[3].missing_value, Nans.NOT_MISSING)
+    self.assertEqual(rows[3].missing_stderr, Nans.NOT_MISSING)
+    self.assertEqual(rows[3].missing_sample_size, Nans.NOT_MISSING)