Acquisition: refactor csv_importer for cleanliness

* simplify and unify CovidcastRow creation from file
* convert CSV file details tuple to NamedTuple

Author: dshemetov

src/acquisition/covidcast/csv_importer.py

@@ -6,6 +6,7 @@
 from dataclasses import dataclass
 from datetime import date
 from glob import glob
+from typing import Iterator, NamedTuple, Optional, Tuple
 
 # third party
 import epiweeks as epi
@@ -14,6 +15,10 @@
 # first party
 from delphi_utils import Nans
 from delphi.utils.epiweek import delta_epiweeks
+from delphi.epidata.acquisition.covidcast.database import CovidcastRow
+from delphi.epidata.acquisition.covidcast.logger import get_structured_logger
+
+DFRow = NamedTuple('DFRow', [('geo_id', str), ('value', float), ('stderr', float), ('sample_size', float), ('missing_value', int), ('missing_stderr', int), ('missing_sample_size', int)])
 PathDetails = NamedTuple('PathDetails', [('source', str), ("signal", str), ('time_type', str), ('geo_type', str), ('time_value', int), ('issue', int), ('lag', int)])
 
 
@@ -180,7 +185,8 @@ def find_csv_files(scan_dir, issue=(date.today(), epi.Week.fromdate(date.today()
         yield (path, None)
         continue
 
-      yield (path, (source, signal, time_type, geo_type, time_value, issue_value, lag_value))
+      yield (path, PathDetails(source, signal, time_type, geo_type, time_value, issue_value, lag_value))
+
 
   @staticmethod
   def is_header_valid(columns):
@@ -344,7 +350,7 @@ def extract_and_check_row(row: DFRow, geo_type: str, filepath: Optional[str] = N
 
 
   @staticmethod
-  def load_csv(filepath, geo_type):
+  def load_csv(filepath: str, details: PathDetails) -> Iterator[Optional[CovidcastRow]]:
     """Load, validate, and yield data as `RowValues` from a CSV file.
 
     filepath: the CSV file to be loaded
@@ -369,9 +375,32 @@ def load_csv(filepath, geo_type):
     table.rename(columns={"val": "value", "se": "stderr", "missing_val": "missing_value", "missing_se": "missing_stderr"}, inplace=True)
 
     for row in table.itertuples(index=False):
-      row_values, error = CsvImporter.extract_and_check_row(row, geo_type, filepath)
+      csv_row_values, error = CsvImporter.extract_and_check_row(row, details.geo_type, filepath)
+
       if error:
         logger.warning(event = 'invalid value for row', detail=(str(row), error), file=filepath)
         yield None
         continue
-      yield row_values
+
+      yield CovidcastRow(
+        details.source,
+        details.signal,
+        details.time_type,
+        details.geo_type,
+        details.time_value,
+        csv_row_values.geo_value,
+        csv_row_values.value,
+        csv_row_values.stderr,
+        csv_row_values.sample_size,
+        csv_row_values.missing_value,
+        csv_row_values.missing_stderr,
+        csv_row_values.missing_sample_size,
+        details.issue,
+        details.lag,
+        # These four fields are unused by database acquisition
+        # TODO: These will be used when CovidcastRow is updated.
+        # id=None,
+        # direction=None,
+        # direction_updated_timestamp=0,
+        # value_updated_timestamp=0,
+      )

src/acquisition/covidcast/csv_to_database.py

@@ -103,12 +103,8 @@ def upload_archive(
       archive_as_failed(path_src, filename, 'unknown',logger)
       continue
 
-    (source, signal, time_type, geo_type, time_value, issue, lag) = details
-
-    csv_rows = csv_importer_impl.load_csv(path, geo_type)
-
-    cc_rows = CovidcastRow.fromCsvRows(csv_rows, source, signal, time_type, geo_type, time_value, issue, lag)
-    rows_list = list(cc_rows)
+    csv_rows = CsvImporter.load_csv(path, details)
+    rows_list = list(csv_rows)
     all_rows_valid = rows_list and all(r is not None for r in rows_list)
     if all_rows_valid:
       try:
@@ -117,12 +113,13 @@ def upload_archive(
         logger.info(
           "Inserted database rows",
           row_count = modified_row_count,
-          source = source,
-          signal = signal,
-          geo_type = geo_type,
-          time_value = time_value,
-          issue = issue,
-          lag = lag)
+          source = details.source,
+          signal = details.signal,
+          geo_type = details.geo_type,
+          time_value = details.time_value,
+          issue = details.issue,
+          lag = details.lag
+        )
         if modified_row_count is None or modified_row_count: # else would indicate zero rows inserted
           total_modified_row_count += (modified_row_count if modified_row_count else 0)
           database.commit()
@@ -137,9 +134,9 @@ def upload_archive(
 
     # archive the current file based on validation results
     if all_rows_valid:
-      archive_as_successful(path_src, filename, source, logger)
+      archive_as_successful(path_src, filename, details.source, logger)
     else:
-      archive_as_failed(path_src, filename, source,logger)
+      archive_as_failed(path_src, filename, details.source, logger)
 
   return total_modified_row_count
 

tests/acquisition/covidcast/test_csv_importer.py

@@ -233,10 +233,10 @@ def test_load_csv_with_invalid_header(self, mock_read_csv):
 
     data = {'foo': [1, 2, 3]}
     filepath = 'path/name.csv'
-    geo_type = 'state'
+    details = PathDetails("src", "name", "day", "state", 20200101, 20200101, 0)
 
     mock_read_csv.return_value = pd.DataFrame(data)
-    rows = list(CsvImporter.load_csv(filepath, geo_type))
+    rows = list(CsvImporter.load_csv(filepath, details))
 
     self.assertTrue(mock_read_csv.called)
     self.assertTrue(mock_read_csv.call_args[0][0], filepath)
@@ -255,10 +255,10 @@ def test_load_csv_with_valid_header(self, mock_read_csv):
       'sample_size': ['301', '302', '303', '304'],
     }
     filepath = 'path/name.csv'
-    geo_type = 'state'
+    details = PathDetails("src", "name", "day", "state", 20200101, 20200101, 0)
 
     mock_read_csv.return_value = pd.DataFrame(data=data)
-    rows = list(CsvImporter.load_csv(filepath, geo_type))
+    rows = list(CsvImporter.load_csv(filepath, details))
 
     self.assertTrue(mock_read_csv.called)
     self.assertTrue(mock_read_csv.call_args[0][0], filepath)
@@ -292,10 +292,10 @@ def test_load_csv_with_valid_header(self, mock_read_csv):
       'missing_sample_size': [Nans.NOT_MISSING] * 2 + [Nans.REGION_EXCEPTION] * 2 + [None]
     }
     filepath = 'path/name.csv'
-    geo_type = 'state'
+    details = PathDetails("src", "name", "day", "state", 20200101, 20200101, 0)
 
     mock_read_csv.return_value = pd.DataFrame(data)
-    rows = list(CsvImporter.load_csv(filepath, geo_type))
+    rows = list(CsvImporter.load_csv(filepath, details))
 
     self.assertTrue(mock_read_csv.called)
     self.assertTrue(mock_read_csv.call_args[0][0], filepath)

tests/acquisition/covidcast/test_csv_to_database.py

@@ -15,6 +15,14 @@
 
 class UnitTests(unittest.TestCase):
   """Basic unit tests."""
+  _path_details = [
+    # a good file
+    ('path/a.csv', PathDetails('src_a', 'sig_a', 'day', 'hrr', 20200419, 20200420, 1)),
+    # a file with a data error
+    ('path/b.csv', PathDetails('src_b', 'sig_b', 'week', 'msa', 202016, 202017, 1)),
+    # emulate a file that's named incorrectly
+    ('path/c.csv', None)
+  ]
 
   def test_get_argument_parser(self):
     """Return a parser for command-line arguments."""
@@ -37,25 +45,32 @@ def test_collect_files(self, mock_csv_importer: MagicMock):
   def test_upload_archive(self, mock_file_archiver: MagicMock, mock_csv_importer: MagicMock, mock_database: MagicMock):
     """Upload to the database, and archive."""
 
-    def make_row(value):
+    def make_row(value: float, details: PathDetails):
       return MagicMock(
+        source=details.source,
+        signal=details.signal,
+        time_type=details.time_type,
+        geo_type=details.geo_type,
+        time_value=details.time_value,
+        issue=details.issue,
+        lag=details.lag,
         geo_value=value,
         value=value,
         stderr=value,
         sample_size=value,
       )
 
-    def load_csv_impl(path, *args):
+    def load_csv_impl(path, details):
       if path == 'path/a.csv':
         # no validation errors
-        yield make_row('a1')
-        yield make_row('a2')
-        yield make_row('a3')
+        yield make_row('a1', details)
+        yield make_row('a2', details)
+        yield make_row('a3', details)
       elif path == 'path/b.csv':
         # one validation error
-        yield make_row('b1')
+        yield make_row('b1', details)
         yield None
-        yield make_row('b3')
+        yield make_row('b3', details)
       else:
         # fail the test for any other path
         raise Exception('unexpected path')
@@ -69,7 +84,7 @@ def iter_len(l: Iterable) -> int:
     mock_logger = MagicMock()
 
     modified_row_count = upload_archive(
-      self._path_details(),
+      self._path_details,
       mock_database,
       make_handlers(data_dir, False),
       mock_logger
@@ -179,7 +194,7 @@ def test_database_exception_is_handled(self, mock_file_archiver: MagicMock, mock
     data_dir = 'data_dir'
     mock_database.insert_or_update_bulk.side_effect = Exception('testing')
     mock_csv_importer.find_csv_files.return_value = [
-      ('path/file.csv', ('src', 'sig', 'day', 'hrr', 20200423, 20200424, 1)),
+      ('path/file.csv', PathDetails('src', 'sig', 'day', 'hrr', 20200423, 20200424, 1)),
     ]
     mock_csv_importer.load_csv.return_value = [
       MagicMock(geo_value='geo', value=1, stderr=1, sample_size=1),