use issue specific CSV files

Author: chinandrew

integrations/acquisition/covidcast_nowcast/test_csv_uploading.py

@@ -68,15 +68,15 @@ def test_uploading(self):
     # print full diff if something unexpected comes out
     self.maxDiff=None
 
-    receiving_dir = '/common/covidcast_nowcast/receiving/src/'
-    success_dir = '/common/covidcast_nowcast/archive/successful/src/'
-    failed_dir = '/common/covidcast_nowcast/archive/failed/src/'
+    receiving_dir = '/common/covidcast_nowcast/receiving/issue_20200421/src/'
+    success_dir = '/common/covidcast_nowcast/archive/successful/issue_20200421/src/'
+    failed_dir = '/common/covidcast_nowcast/archive/failed/issue_20200421/src/'
     os.makedirs(receiving_dir, exist_ok=True)
 
     # valid
     with open(receiving_dir + '20200419_state_sig.csv', 'w') as f:
-      f.write('sensor_name,geo_value,value,issue\n')
-      f.write('testsensor,ca,1,20200421\n')
+      f.write('sensor_name,geo_value,value\n')
+      f.write('testsensor,ca,1\n')
 
     # invalid filename
     with open(receiving_dir + 'hello.csv', 'w') as f:
@@ -125,16 +125,16 @@ def test_duplicate_row(self):
     # print full diff if something unexpected comes out
     self.maxDiff=None
 
-    receiving_dir = '/common/covidcast_nowcast/receiving/src/'
+    receiving_dir = '/common/covidcast_nowcast/receiving/issue_20200425/src/'
     os.makedirs(receiving_dir, exist_ok=True)
 
     with open(receiving_dir + '20200419_state_sig.csv', 'w') as f:
-      f.write('sensor_name,geo_value,value,issue\n')
-      f.write('testsensor,ca,1,20200425\n')
+      f.write('sensor_name,geo_value,value\n')
+      f.write('testsensor,ca,1\n')
     main()
     with open(receiving_dir + '20200419_state_sig.csv', 'w') as f:
-      f.write('sensor_name,geo_value,value,issue\n')
-      f.write('testsensor,ca,2,20200425\n')
+      f.write('sensor_name,geo_value,value\n')
+      f.write('testsensor,ca,2\n')
     main()
 
     # most most recent value is the one stored

src/acquisition/covidcast_nowcast/load_sensors.py

@@ -13,7 +13,7 @@
 FAIL_DIR = "archive/failed"
 TABLE_NAME = "covidcast_nowcast"
 DB_NAME = "epidata"
-CSV_DTYPES = {"sensor_name": str, "geo_value": str, "value": float, "issue": int}
+CSV_DTYPES = {"sensor_name": str, "geo_value": str, "value": float}
 
 
 def main(csv_path: str = SENSOR_CSV_PATH) -> None:
@@ -37,7 +37,8 @@ def main(csv_path: str = SENSOR_CSV_PATH) -> None:
     """
     user, pw = secrets.db.epi
     engine = sqlalchemy.create_engine(f"mysql+pymysql://{user}:{pw}@{secrets.db.host}/{DB_NAME}")
-    for filepath, attribute in CsvImporter.find_csv_files(csv_path):
+#    for filepath, attribute in CsvImporter.find_csv_files(csv_path):
+    for filepath, attribute in CsvImporter.find_issue_specific_csv_files(csv_path):
         if attribute is None:
             _move_after_processing(filepath, success=False)
             continue
@@ -76,8 +77,8 @@ def load_and_prepare_file(filepath: str, attributes: tuple) -> pd.DataFrame:
     data["time_type"] = time_type
     data["geo_type"] = geo_type
     data["time_value"] = time_value
-    data["lag"] = [(datetime.strptime(str(i), "%Y%m%d") - datetime.strptime(str(j), "%Y%m%d")).days
-                   for i, j in zip(data["issue"], data["time_value"])]
+    data["issue"] = issue_value
+    data["lag"] = lag_value
     data["value_updated_timestamp"] = int(time.time())
     return data
 

tests/acquisition/covidcast_nowcast/test_load_sensors.py

@@ -25,20 +25,20 @@ def test_load_and_prepare_file(self):
                        "test_time_type",
                        "test_geo_type",
                        20201231,
-                       "test_issue_value",
-                       "test_lag_value")
+                       20210102,
+                       3)
 
-    test_df = load_and_prepare_file(StringIO("sensor_name,geo_value,value,issue\ntestname,01001,1.5,20210102"), test_attributes)
+    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,
                                   pd.DataFrame({"sensor_name": ["testname"],
                                                 "geo_value": ["01001"],
                                                 "value": [1.5],
-                                                "issue": [20210102],
                                                 "source": ["test_source"],
                                                 "signal": ["test_signal"],
                                                 "time_type": ["test_time_type"],
                                                 "geo_type": ["test_geo_type"],
                                                 "time_value": [20201231],
-                                                "lag": [2],
+                                                "issue": [20210102],
+                                                "lag": [3],
                                                 "value_updated_timestamp": [12345]})
                                   )