Merge pull request #992 from cmu-delphi/nowcast-ground-truth

Nowcasting ground truth and refactor

Author: krivard

nowcast/delphi_nowcast/deconvolution/deconvolution.py

@@ -0,0 +1,6 @@
+"""Deconvolution functions."""
+
+
+def deconvolve_double_smooth_tf_cv(x, y, z):
+    x, y, z
+    pass

nowcast/delphi_nowcast/deconvolution/ground_truth.py

@@ -0,0 +1,47 @@
+"""Functions to construct the ground truth for a given location."""
+from datetime import date
+from typing import List, Dict, Tuple
+
+import numpy as np
+
+from .deconvolution import deconvolve_double_smooth_tf_cv
+from ..constants import Default
+from ..data_containers import LocationSeries, SensorConfig
+from ..epidata import get_historical_sensor_data, get_indicator_data, export_to_csv
+
+
+def construct_truths(start_date: date,
+                     end_date: date,
+                     as_of: date,  # most likely today
+                     truth: SensorConfig,
+                     locations: List[LocationSeries],
+                     export_dir: str = "") -> Dict[Tuple, LocationSeries]:
+    raw_indicator = get_indicator_data([truth], locations, as_of)
+    output = {}
+    for location in locations:
+        indicator_key = (truth.source, truth.signal, location.geo_type, location.geo_value)
+        location, missing_dates = get_historical_sensor_data(truth, location, end_date, start_date)
+        location, export = fill_missing_days(location, raw_indicator[indicator_key], missing_dates)
+        output[indicator_key] = location
+        if export_dir and export.values:
+            export_to_csv(export,  truth, as_of, export_dir)
+        return output
+
+
+def fill_missing_days(stored_vals: LocationSeries,
+                      indicator_data: LocationSeries,
+                      missing_dates: List[date]):
+    export_data = LocationSeries(stored_vals.geo_value, stored_vals.geo_type)
+    for day in missing_dates:
+        try:
+            y = np.array(indicator_data.get_data_range(min(indicator_data.dates), day, "linear"))
+            x = np.arange(1, len(y) + 1)
+        except ValueError:
+            continue
+        deconv_vals = deconvolve_double_smooth_tf_cv(
+            y, x, Default.DELAY_DISTRIBUTION
+        )
+        missing_day_val = deconv_vals[-1]
+        stored_vals.add_data(day, missing_day_val)
+        export_data.add_data(day, missing_day_val)  # holds only data to get exported
+    return stored_vals, export_data

nowcast/delphi_nowcast/get_epidata.py renamed to nowcast/delphi_nowcast/epidata.py

@@ -1,4 +1,5 @@
-"""Retrieve data from Epidata API."""
+"""Functions for interfacing with Epidata."""
+import os
 from datetime import datetime, date
 from itertools import product
 from typing import Tuple, List, Dict
@@ -67,8 +68,7 @@ def get_indicator_data(sensors: List[SensorConfig],
 
 
 def get_historical_sensor_data(sensor: SensorConfig,
-                               geo_value: str,
-                               geo_type: str,
+                               location: LocationSeries,
                                start_date: date,
                                end_date: date) -> Tuple[LocationSeries, list]:
     """
@@ -81,10 +81,8 @@ def get_historical_sensor_data(sensor: SensorConfig,
     ----------
     sensor
         SensorConfig specifying which sensor to retrieve.
-    geo_type
-        Geo type to retrieve.
-    geo_value
-        Geo value to retrieve.
+    location
+        LocationSeries for the location to get.
     start_date
         First day to retrieve (inclusive).
     end_date
@@ -98,23 +96,65 @@ def get_historical_sensor_data(sensor: SensorConfig,
         data_source=sensor.source,
         signals=sensor.signal,
         time_type="day",
-        geo_type=geo_type,
+        geo_type=location.geo_type,
         time_values=Epidata.range(start_date.strftime("%Y%m%d"), end_date.strftime("%Y%m%d")),
-        geo_value=geo_value,
+        geo_value=location.geo_value,
         sensor_names=sensor.name,
         lag=sensor.lag)
     all_dates = [i.date() for i in date_range(start_date, end_date)]
     if response["result"] == 1:
-        output = LocationSeries(
-            geo_value=geo_value,
-            geo_type=geo_type,
-            data={datetime.strptime(str(i["time_value"]), "%Y%m%d").date(): i["value"]
-                  for i in response.get("epidata", []) if not isnan(i["value"])}
-        )
-        missing_dates = [i for i in all_dates if i not in output.dates]
-        return output, missing_dates
+        location.data = {datetime.strptime(str(i["time_value"]), "%Y%m%d").date(): i["value"]
+                         for i in response.get("epidata", []) if not isnan(i["value"])}
+        missing_dates = [i for i in all_dates if i not in location.dates]
+        return location, missing_dates
     if response["result"] == -2:  # no results
         print("No historical results found")
-        output = LocationSeries(geo_value=geo_value, geo_type=geo_type)
-        return output, all_dates
+        return location, all_dates
     raise Exception(f"Bad result from Epidata: {response['message']}")
+
+
+def export_to_csv(value: LocationSeries,
+                   sensor: SensorConfig,
+                   as_of_date: date,
+                   receiving_dir: str
+                   ) -> List[str]:
+    """
+    Save value to csv for upload to Epidata database.
+
+    Parameters
+    ----------
+    value
+        LocationSeries containing data.
+    sensor
+        SensorConfig corresponding to value.
+    as_of_date
+        As_of date for the indicator data used to train the sensor.
+    receiving_dir
+        Export directory for Epidata acquisition.
+    Returns
+    -------
+        Filepath of exported files
+    """
+    export_dir = os.path.join(
+        receiving_dir,
+        f"issue_{as_of_date.strftime('%Y%m%d')}",
+        sensor.source
+    )
+    os.makedirs(export_dir, exist_ok=True)
+    exported_files = []
+    for time_value in value.dates:
+        export_file = os.path.join(
+            export_dir,
+            f"{time_value.strftime('%Y%m%d')}_{value.geo_type}_{sensor.signal}.csv"
+        )
+        if os.path.exists(export_file):
+            with open(export_file, "a") as f:
+                f.write(
+                    f"{sensor.name},{value.geo_value},{value.data.get(time_value, '')}\n")
+        else:
+            with open(export_file, "a") as f:
+                f.write("sensor_name,geo_value,value\n")
+                f.write(
+                    f"{sensor.name},{value.geo_value},{value.data.get(time_value, '')}\n")
+        exported_files.append(export_file)
+    return exported_files

nowcast/delphi_nowcast/sensorization/sensor.py

@@ -8,9 +8,9 @@
 
 from .ar_model import compute_ar_sensor
 from .regression_model import compute_regression_sensor
-from ..get_epidata import get_indicator_data, get_historical_sensor_data
 from ..data_containers import LocationSeries, SensorConfig
 from ..constants import AR_ORDER, AR_LAMBDA, REG_INTERCEPT
+from ..epidata import get_indicator_data, get_historical_sensor_data, export_to_csv
 
 
 def compute_sensors(as_of_date: date,
@@ -65,7 +65,7 @@ def compute_sensors(as_of_date: date,
     if export_dir:
         for sensor, locations in output.items():
             for loc in locations:
-                print(_export_to_csv(loc, sensor, as_of_date, export_dir))
+                print(export_to_csv(loc, sensor, as_of_date, export_dir))
     return output
 
 
@@ -96,54 +96,8 @@ def historical_sensors(start_date: date,
     for location in ground_truths:
         for sensor in sensors:
             sensor_vals, missing_dates = get_historical_sensor_data(
-                sensor, location.geo_value, location.geo_type, start_date, end_date
+                sensor, location, start_date, end_date
             )
             if sensor_vals.data:
                 output[sensor].append(sensor_vals)
     return output
-
-
-def _export_to_csv(value: LocationSeries,
-                   sensor: SensorConfig,
-                   as_of_date: date,
-                   receiving_dir: str
-                   ) -> List[str]:
-    """
-    Save value to csv for upload to Epidata database.
-    Parameters
-    ----------
-    value
-        LocationSeries containing data.
-    sensor
-        SensorConfig corresponding to value.
-    as_of_date
-        As_of date for the indicator data used to train the sensor.
-    receiving_dir
-        Export directory for Epidata acquisition.
-    Returns
-    -------
-        Filepath of exported files
-    """
-    export_dir = os.path.join(
-        receiving_dir,
-        f"issue_{as_of_date.strftime('%Y%m%d')}",
-        sensor.source
-    )
-    os.makedirs(export_dir, exist_ok=True)
-    exported_files = []
-    for time_value in value.dates:
-        export_file = os.path.join(
-            export_dir,
-            f"{time_value.strftime('%Y%m%d')}_{value.geo_type}_{sensor.signal}.csv"
-        )
-        if os.path.exists(export_file):
-            with open(export_file, "a") as f:
-                f.write(
-                    f"{sensor.name},{value.geo_value},{value.data.get(time_value, '')}\n")
-        else:
-            with open(export_file, "a") as f:
-                f.write("sensor_name,geo_value,value\n")
-                f.write(
-                    f"{sensor.name},{value.geo_value},{value.data.get(time_value, '')}\n")
-        exported_files.append(export_file)
-    return exported_files

nowcast/tests/sensorization/test_sensor.py

@@ -8,7 +8,7 @@
 import pandas as pd
 
 from delphi_nowcast.data_containers import LocationSeries, SensorConfig
-from delphi_nowcast.sensorization.sensor import compute_sensors, historical_sensors, _export_to_csv
+from delphi_nowcast.sensorization.sensor import compute_sensors, historical_sensors
 
 
 class TestComputeSensors:
@@ -72,25 +72,3 @@ def test_historical_sensors_no_data(self, mock_historical):
         test_ground_truth = [LocationSeries("ca", "state")]
         assert historical_sensors(
             None, None, test_sensors, test_ground_truth) == {}
-
-
-class TestExportToCSV:
-
-    def test__export_to_csv(self):
-        """Test export creates the right file and right contents."""
-        test_sensor = SensorConfig(source="src",
-                                   signal="sig",
-                                   name="test",
-                                   lag=4)
-        test_value = LocationSeries("ca", "state", {date(2020, 1, 1): 1.5})
-        with tempfile.TemporaryDirectory() as tmpdir:
-            out_files = _export_to_csv(test_value, test_sensor, date(2020, 1, 5), receiving_dir=tmpdir)
-            assert len(out_files) == 1
-            out_file = out_files[0]
-            assert os.path.isfile(out_file)
-            assert out_file.endswith("issue_20200105/src/20200101_state_sig.csv")
-            out_file_df = pd.read_csv(out_file)
-            pd.testing.assert_frame_equal(out_file_df,
-                                          pd.DataFrame({"sensor_name": ["test"],
-                                                        "geo_value": ["ca"],
-                                                        "value": [1.5]}))

nowcast/tests/test_get_epidata.py renamed to nowcast/tests/test_epidata.py

@@ -1,12 +1,13 @@
-from unittest.mock import patch
+import os
+import tempfile
 from datetime import date
+from unittest.mock import patch
 
-import pytest
 import numpy as np
-
-from delphi_nowcast.get_epidata import get_indicator_data, get_historical_sensor_data, \
-    EPIDATA_START_DATE
+import pandas as pd
+import pytest
 from delphi_nowcast.data_containers import LocationSeries, SensorConfig
+from delphi_nowcast.epidata import export_to_csv, get_indicator_data, get_historical_sensor_data, EPIDATA_START_DATE
 
 
 class TestGetIndicatorData:
@@ -130,8 +131,7 @@ def test_results(self, mock_epidata):
                         {"time_value": 20200102, "value": np.nan}]
         }
         test_output = get_historical_sensor_data(SensorConfig(None, None, None, None),
-                                                 None,
-                                                 None,
+                                                 LocationSeries(None, None),
                                                  date(2020, 1, 1),
                                                  date(2020, 1, 4))
 
@@ -144,8 +144,7 @@ def test_results(self, mock_epidata):
     def test_no_results(self, mock_epidata):
         mock_epidata.return_value = {"result": -2}
         test_output = get_historical_sensor_data(SensorConfig(None, None, None, None),
-                                                 None,
-                                                 None,
+                                                 LocationSeries(None, None),
                                                  date(2020, 1, 1),
                                                  date(2020, 1, 4))
 
@@ -157,7 +156,28 @@ def test_error(self, mock_epidata):
         mock_epidata.return_value = {"result": -3, "message": "test failure"}
         with pytest.raises(Exception, match="Bad result from Epidata: test failure"):
             get_historical_sensor_data(SensorConfig(None, None, None, None),
-                                       None,
-                                       None,
+                                       LocationSeries(None, None),
                                        date(2020, 1, 1),
                                        date(2020, 1, 4))
+
+
+class TestExportToCSV:
+
+    def test_export_to_csv(self):
+        """Test export creates the right file and right contents."""
+        test_sensor = SensorConfig(source="src",
+                                   signal="sig",
+                                   name="test",
+                                   lag=4)
+        test_value = LocationSeries("ca", "state", {date(2020, 1, 1): 1.5})
+        with tempfile.TemporaryDirectory() as tmpdir:
+            out_files = export_to_csv(test_value, test_sensor, date(2020, 1, 5), receiving_dir=tmpdir)
+            assert len(out_files) == 1
+            out_file = out_files[0]
+            assert os.path.isfile(out_file)
+            assert out_file.endswith("issue_20200105/src/20200101_state_sig.csv")
+            out_file_df = pd.read_csv(out_file)
+            pd.testing.assert_frame_equal(out_file_df,
+                                          pd.DataFrame({"sensor_name": ["test"],
+                                                        "geo_value": ["ca"],
+                                                        "value": [1.5]}))