Acquisition: change PathDetails order to path order

* update covidcast_nowcast
* update tests

Author: dshemetov

src/acquisition/covidcast/csv_importer.py

@@ -19,7 +19,7 @@
 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)])
+PathDetails = NamedTuple('PathDetails', [('issue', int), ('lag', int), ('source', str), ('signal', str), ('time_type', str), ('time_value', int), ('geo_type', str)])
 
 
 @dataclass
@@ -185,7 +185,7 @@ def find_csv_files(scan_dir, issue=(date.today(), epi.Week.fromdate(date.today()
         yield (path, None)
         continue
 
-      yield (path, PathDetails(source, signal, time_type, geo_type, time_value, issue_value, lag_value))
+      yield (path, PathDetails(issue_value, lag_value, source, signal, time_type, time_value, geo_type))
 
 
   @staticmethod

src/acquisition/covidcast_nowcast/load_sensors.py

@@ -6,7 +6,7 @@
 import sqlalchemy
 
 import delphi.operations.secrets as secrets
-from delphi.epidata.acquisition.covidcast.csv_importer import CsvImporter
+from delphi.epidata.acquisition.covidcast.csv_importer import CsvImporter, PathDetails
 
 SENSOR_CSV_PATH = "/common/covidcast_nowcast/receiving/"
 SUCCESS_DIR = "archive/successful"
@@ -52,7 +52,7 @@ def main(csv_path: str = SENSOR_CSV_PATH) -> None:
         _move_after_processing(filepath, success=True)
 
 
-def load_and_prepare_file(filepath: str, attributes: tuple) -> pd.DataFrame:
+def load_and_prepare_file(filepath: str, attributes: PathDetails) -> pd.DataFrame:
     """
     Read CSV file into a DataFrame and add relevant attributes as new columns to match DB table.
 
@@ -68,15 +68,14 @@ def load_and_prepare_file(filepath: str, attributes: tuple) -> pd.DataFrame:
     -------
         DataFrame with additional attributes added as columns based on filename and current date.
     """
-    source, signal, time_type, geo_type, time_value, issue_value, lag_value = attributes
     data = pd.read_csv(filepath, dtype=CSV_DTYPES)
-    data["source"] = source
-    data["signal"] = signal
-    data["time_type"] = time_type
-    data["geo_type"] = geo_type
-    data["time_value"] = time_value
-    data["issue"] = issue_value
-    data["lag"] = lag_value
+    data["source"] = attributes.source
+    data["signal"] = attributes.signal
+    data["time_type"] = attributes.time_type
+    data["geo_type"] = attributes.geo_type
+    data["time_value"] = attributes.time_value
+    data["issue"] = attributes.issue
+    data["lag"] = attributes.lag
     data["value_updated_timestamp"] = int(time.time())
     return data
 

tests/acquisition/covidcast/test_csv_importer.py

@@ -105,10 +105,10 @@ def test_find_csv_files(self, mock_glob: MagicMock):
     expected_issue_week=int(str(epi.Week.fromdate(date.today())))
     time_value_day = 20200408
     expected = set([
-      (glob_paths[0], ('fb_survey', 'cli', 'week', 'county', 202015, expected_issue_week, delta_epiweeks(202015, expected_issue_week))),
-      (glob_paths[1], ('ght', 'rawsearch', 'day', 'state', time_value_day, expected_issue_day, (date.today() - date(year=time_value_day // 10000, month=(time_value_day // 100) % 100, day=time_value_day % 100)).days)),
-      (glob_paths[2], ('valid', 'sig', 'day', 'nation', time_value_day, expected_issue_day, (date.today() - date(year=time_value_day // 10000, month=(time_value_day // 100) % 100, day=time_value_day % 100)).days)),
-      (glob_paths[3], ('valid', 'sig', 'day', 'hhs', time_value_day, expected_issue_day, (date.today() - date(year=time_value_day // 10000, month=(time_value_day // 100) % 100, day=time_value_day % 100)).days)),
+      (glob_paths[0], PathDetails(expected_issue_week, delta_epiweeks(202015, expected_issue_week), 'fb_survey', 'cli', 'week', 202015, 'county')),
+      (glob_paths[1], PathDetails(expected_issue_day, (date.today() - date(year=time_value_day // 10000, month=(time_value_day // 100) % 100, day=time_value_day % 100)).days, 'ght', 'rawsearch', 'day', time_value_day, 'state')),
+      (glob_paths[2], PathDetails(expected_issue_day, (date.today() - date(year=time_value_day // 10000, month=(time_value_day // 100) % 100, day=time_value_day % 100)).days, 'valid', 'sig', 'day', time_value_day, 'nation')),
+      (glob_paths[3], PathDetails(expected_issue_day, (date.today() - date(year=time_value_day // 10000, month=(time_value_day // 100) % 100, day=time_value_day % 100)).days, 'valid', 'sig', 'day', time_value_day, 'hhs')),
       (glob_paths[4], None),
       (glob_paths[5], None),
       (glob_paths[6], None),
@@ -233,7 +233,7 @@ def test_load_csv_with_invalid_header(self, mock_read_csv):
 
     data = {'foo': [1, 2, 3]}
     filepath = 'path/name.csv'
-    details = PathDetails("src", "name", "day", "state", 20200101, 20200101, 0)
+    details = PathDetails(20200101, 0, "src", "name", "day", 20200101, "state")
 
     mock_read_csv.return_value = pd.DataFrame(data)
     rows = list(CsvImporter.load_csv(filepath, details))
@@ -255,7 +255,7 @@ def test_load_csv_with_valid_header(self, mock_read_csv):
       'sample_size': ['301', '302', '303', '304'],
     }
     filepath = 'path/name.csv'
-    details = PathDetails("src", "name", "day", "state", 20200101, 20200101, 0)
+    details = PathDetails(20200101, 0, "src", "name", "day", 20200101, "state")
 
     mock_read_csv.return_value = pd.DataFrame(data=data)
     rows = list(CsvImporter.load_csv(filepath, details))
@@ -292,7 +292,7 @@ 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'
-    details = PathDetails("src", "name", "day", "state", 20200101, 20200101, 0)
+    details = PathDetails(20200101, 0, "src", "name", "day", 20200101, "state")
 
     mock_read_csv.return_value = pd.DataFrame(data)
     rows = list(CsvImporter.load_csv(filepath, details))

tests/acquisition/covidcast/test_csv_to_database.py

@@ -17,9 +17,9 @@ 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)),
+    ('path/a.csv', PathDetails(20200420, 1, 'src_a', 'sig_a', 'day', 20200419, 'hrr')),
     # a file with a data error
-    ('path/b.csv', PathDetails('src_b', 'sig_b', 'week', 'msa', 202016, 202017, 1)),
+    ('path/b.csv', PathDetails(202017, 1, 'src_b', 'sig_b', 'week', 202016, 'msa')),
     # emulate a file that's named incorrectly
     ('path/c.csv', None)
   ]
@@ -194,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', PathDetails('src', 'sig', 'day', 'hrr', 20200423, 20200424, 1)),
+      ('path/file.csv', PathDetails(20200424, 1, 'src', 'sig', 'day', 20200423, 'hrr')),
     ]
     mock_csv_importer.load_csv.return_value = [
       MagicMock(geo_value='geo', value=1, stderr=1, sample_size=1),

tests/acquisition/covidcast_nowcast/test_load_sensors.py

@@ -9,6 +9,7 @@
 import pandas as pd
 
 # first party
+from delphi.epidata.acquisition.covidcast.csv_importer import PathDetails
 from delphi.epidata.acquisition.covidcast_nowcast.load_sensors import main, load_and_prepare_file
 
 # py3tester coverage target
@@ -20,13 +21,15 @@ class UpdateTests(unittest.TestCase):
   @mock.patch('time.time', mock.MagicMock(return_value=12345))
   def test_load_and_prepare_file(self):
 
-    test_attributes = ("test_source",
-                       "test_signal",
-                       "test_time_type",
-                       "test_geo_type",
-                       20201231,
-                       20210102,
-                       3)
+    test_attributes = PathDetails(
+      20210102,
+      3,
+      "test_source",
+      "test_signal",
+      "test_time_type",
+      20201231,
+      "test_geo_type",
+    )
 
     test_df = load_and_prepare_file(StringIO("sensor_name,geo_value,value\ntestname,01001,1.5"), test_attributes)
     pd.testing.assert_frame_equal(test_df,