Add lag calculation

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

src/acquisition/covidcast_nowcast/load_sensors.py

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

tests/acquisition/covidcast_nowcast/test_load_sensors.py

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