use lag and issue from csv file

Author: chinandrew

integrations/acquisition/covidcast_nowcast/test_csv_uploading.py

@@ -75,8 +75,8 @@ def test_uploading(self):
 
     # valid
     with open(receiving_dir + '20200419_state_sig.csv', 'w') as f:
-      f.write('sensor_name,geo_value,value\n')
-      f.write('testsensor,ca,1\n')
+      f.write('sensor_name,geo_value,value,lag,issue\n')
+      f.write('testsensor,ca,1,2,20200421\n')
 
     # invalid filename
     with open(receiving_dir + 'hello.csv', 'w') as f:
@@ -129,12 +129,12 @@ def test_duplicate_row(self):
     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\n')
-      f.write('testsensor,ca,1\n')
+      f.write('sensor_name,geo_value,value,lag,issue\n')
+      f.write('testsensor,ca,1,2,20200415\n')
     main()
     with open(receiving_dir + '20200419_state_sig.csv', 'w') as f:
-      f.write('sensor_name,geo_value,value\n')
-      f.write('testsensor,ca,2\n')
+      f.write('sensor_name,geo_value,value,lag,issue\n')
+      f.write('testsensor,ca,2,2,20200415\n')
     main()
 
     # most most recent value is the one stored
@@ -146,7 +146,7 @@ def test_duplicate_row(self):
         'time_value': 20200419,
         'geo_value': 'ca',
         'value': 2,
-        'issue': 20200421,
+        'issue': 20200415,
         'lag': 2,
         'signal': 'sig',
       }],

src/acquisition/covidcast_nowcast/load_sensors.py

@@ -12,7 +12,7 @@
 FAIL_DIR = "archive/failed"
 TABLE_NAME = "covidcast_nowcast"
 DB_NAME = "epidata"
-CSV_DTYPES = {"sensor_name": str, "geo_value": str, "value": float}
+CSV_DTYPES = {"sensor_name": str, "geo_value": str, "value": float, "lag": int, "issue": int}
 
 
 def main(csv_path: str = SENSOR_CSV_PATH) -> None:
@@ -75,8 +75,7 @@ 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["issue"] = issue_value
-    data["lag"] = lag_value
+    # we don't use the lag and issue calculation since it's specified in the data.
     data["value_updated_timestamp"] = int(time.time())
     return data
 

tests/acquisition/covidcast_nowcast/test_load_sensors.py

@@ -28,17 +28,17 @@ def test_load_and_prepare_file(self):
                        "test_issue_value",
                        "test_lag_value")
 
-    test_df = load_and_prepare_file(StringIO("sensor_name,geo_value,value\ntestname,01001,1.5"), test_attributes)
+    test_df = load_and_prepare_file(StringIO("sensor_name,geo_value,value,lag,issue\ntestname,01001,1.5,2,20200101"), test_attributes)
     pd.testing.assert_frame_equal(test_df,
                                   pd.DataFrame({"sensor_name": ["testname"],
                                                 "geo_value": ["01001"],
                                                 "value": [1.5],
+                                                "lag": [2],
+                                                "issue": [20200101],
                                                 "source": ["test_source"],
                                                 "signal": ["test_signal"],
                                                 "time_type": ["test_time_type"],
                                                 "geo_type": ["test_geo_type"],
                                                 "time_value": ["test_time_value"],
-                                                "issue": ["test_issue_value"],
-                                                "lag": ["test_lag_value"],
                                                 "value_updated_timestamp": [12345]})
                                   )