Nans JHU:

* keep nan values, add missing columns, add tests

Author: dshemetov

_delphi_utils_python/delphi_utils/archive.py

@@ -112,6 +112,11 @@ def diff_export_csv(
     deleted_df[["val", "se", "sample_size"]] = np.nan
     deleted_df[["missing_val", "missing_se", "missing_sample_size"]] = Nans.DELETED
 
+    # Code deleted entries as nans with the deleted missing code
+    deleted_df = before_df.loc[deleted_idx, :].copy()
+    deleted_df[["val", "se", "sample_size"]] = np.nan
+    deleted_df[["missing_val", "missing_se", "missing_sample_size"]] = Nans.DELETED
+
     return (
         deleted_df,
         after_df_cmn.loc[~(same_mask.all(axis=1)), :],

jhu/delphi_jhu/run.py

@@ -9,13 +9,15 @@
 import time
 from typing import Dict, Any
 
+import pandas as pd
 import numpy as np
 from delphi_utils import (
     create_export_csv,
     S3ArchiveDiffer,
     Smoother,
     GeoMapper,
     get_structured_logger,
+    Nans,
 )
 
 from .geo import geo_map
@@ -63,6 +65,34 @@
 ]
 
 
+def add_nancodes(df, metric, geo_res, smoother):
+    """Add nancodes to the dataframe."""
+    idx = pd.IndexSlice
+
+    # Default missingness codes
+    df["missing_val"] = Nans.NOT_MISSING
+    df["missing_se"] = Nans.NOT_APPLICABLE
+    df["missing_sample_size"] = Nans.NOT_APPLICABLE
+
+    # Mark early smoothing entries as data insufficient
+    if smoother == "seven_day_average":
+        df.sort_index(inplace=True)
+        min_time_value = df.index.min()[0] + 5 * pd.Timedelta(days=1)
+        df.loc[idx[:min_time_value, :], "missing_val"] = Nans.PRIVACY
+
+    # Mark Puerto Rico county deaths with a region exception code
+    # Search "Puerto Rico" here for details:
+    # https://github.com/CSSEGISandData/COVID-19/tree/master/csse_covid_19_data
+    if metric == "deaths" and geo_res == "county":
+        puerto_rico_fips = ["72" + str(i).zfill(3) for i in range(1, 155)]
+        df.loc[idx[:, puerto_rico_fips], "missing_val"] = Nans.REGION_EXCEPTION
+
+    # Mark any remaining nans with unknown
+    remaining_nans_mask = df["val"].isnull() & (df["missing_val"] == Nans.NOT_MISSING)
+    df.loc[remaining_nans_mask, "missing_val"] = Nans.UNKNOWN
+    return df
+
+
 def run_module(params: Dict[str, Any]):
     """Run the JHU indicator module.
 
@@ -86,8 +116,10 @@ def run_module(params: Dict[str, Any]):
     export_dir = params["common"]["export_dir"]
     base_url = params["indicator"]["base_url"]
     logger = get_structured_logger(
-        __name__, filename=params["common"].get("log_filename"),
-        log_exceptions=params["common"].get("log_exceptions", True))
+        __name__,
+        filename=params["common"].get("log_filename"),
+        log_exceptions=params["common"].get("log_exceptions", True),
+    )
 
     if "archive" in params:
         arch_diff = S3ArchiveDiffer(
@@ -112,16 +144,22 @@ def run_module(params: Dict[str, Any]):
             metric=metric,
             geo_res=geo_res,
             sensor=sensor,
-            smoother=smoother)
+            smoother=smoother,
+        )
         df = dfs[metric]
         # Aggregate to appropriate geographic resolution
         df = geo_map(df, geo_res, sensor)
         df.set_index(["timestamp", "geo_id"], inplace=True)
+
+        # Smooth
         df["val"] = df[sensor].groupby(level=1).transform(SMOOTHERS_MAP[smoother][0])
+
+        # JHU is not a survey data source
         df["se"] = np.nan
         df["sample_size"] = np.nan
-        # Drop early entries where data insufficient for smoothing
-        df = df[~df["val"].isnull()]
+
+        df = add_nancodes(df, metric, geo_res, smoother)
+
         df = df.reset_index()
         sensor_name = SENSOR_NAME_MAP[sensor][0]
         # if (SENSOR_NAME_MAP[sensor][1] or SMOOTHERS_MAP[smoother][2]):
@@ -141,7 +179,8 @@ def run_module(params: Dict[str, Any]):
             if not oldest_final_export_date:
                 oldest_final_export_date = max(exported_csv_dates)
             oldest_final_export_date = min(
-                oldest_final_export_date, max(exported_csv_dates))
+                oldest_final_export_date, max(exported_csv_dates)
+            )
 
     if arch_diff is not None:
         # Diff exports, and make incremental versions
@@ -167,9 +206,13 @@ def run_module(params: Dict[str, Any]):
     formatted_oldest_final_export_date = None
     if oldest_final_export_date:
         max_lag_in_days = (datetime.now() - oldest_final_export_date).days
-        formatted_oldest_final_export_date = oldest_final_export_date.strftime("%Y-%m-%d")
-    logger.info("Completed indicator run",
-        elapsed_time_in_seconds = elapsed_time_in_seconds,
-        csv_export_count = csv_export_count,
-        max_lag_in_days = max_lag_in_days,
-        oldest_final_export_date = formatted_oldest_final_export_date)
+        formatted_oldest_final_export_date = oldest_final_export_date.strftime(
+            "%Y-%m-%d"
+        )
+    logger.info(
+        "Completed indicator run",
+        elapsed_time_in_seconds=elapsed_time_in_seconds,
+        csv_export_count=csv_export_count,
+        max_lag_in_days=max_lag_in_days,
+        oldest_final_export_date=formatted_oldest_final_export_date,
+    )

jhu/tests/test_run.py

@@ -2,12 +2,25 @@
 from os.path import join, basename
 
 import pandas as pd
+import numpy as np
+from delphi_jhu.run import add_nancodes
+from delphi_utils import Nans
+
+def _non_ignored_files_set(directory):
+    """List all files in a directory not preceded by a '.' and store them in a set."""
+    out = {fname for fname in listdir(directory) if not basename(fname).startswith(".")}
+    return out
+
+def _non_ignored_files_set(directory):
+    """List all files in a directory not preceded by a '.' and store them in a set."""
+    out = {fname for fname in listdir(directory) if not basename(fname).startswith(".")}
+    return out
 
 
 class TestRun:
     def test_output_files_exist(self, run_as_module):
 
-        csv_files = [x for x in listdir("receiving") if not basename(x).startswith(".")]
+        csv_files = _non_ignored_files_set("receiving")
 
         dates = [
             "20200303",
@@ -17,29 +30,86 @@ def test_output_files_exist(self, run_as_module):
             "20200307",
             "20200308",
             "20200309",
-            "20200310",
+            "20200310"
         ]
         geos = ["county", "hrr", "msa", "state", "hhs", "nation"]
-        metrics = []
-        for event in ["confirmed", "deaths"]:
-            for smoothing in ["", "_7dav"]:
-                for window in ["incidence", "cumulative"]:
-                    for stat in ["num", "prop"]:
-                        metrics.append(f"{event}{smoothing}_{window}_{stat}")
-
-        expected_files = []
-        for date in dates:
-            for geo in geos:
-                for metric in metrics:
-                    # Can't compute 7dav for first few days of data because of NAs
-                    if date > "20200305" or "7dav" not in metric:
-                        expected_files += [date + "_" + geo + "_" + metric + ".csv"]
-
-        assert set(csv_files) == set(expected_files)
+        signals = ["confirmed", "deaths"]
+        metrics = [
+            "cumulative_num",
+            "cumulative_prop",
+            "incidence_num",
+            "incidence_prop",
+            "7dav_incidence_num",
+            "7dav_incidence_prop",
+            "7dav_cumulative_num",
+            "7dav_cumulative_prop",
+        ]
+
+        expected_files = {
+            date + "_" + geo + "_" + signal + "_" + metric + ".csv"
+            for date in dates
+            for geo in geos
+            for signal in signals
+            for metric in metrics
+        }
+
+        assert csv_files == expected_files
 
     def test_output_file_format(self, run_as_module):
 
         df = pd.read_csv(
             join("receiving", "20200310_state_confirmed_cumulative_num.csv")
         )
-        assert (df.columns.values == ["geo_id", "val", "se", "sample_size"]).all()
+        assert (
+            df.columns.values
+            == [
+                "geo_id",
+                "val",
+                "se",
+                "sample_size",
+                "missing_val",
+                "missing_se",
+                "missing_sample_size",
+            ]
+        ).all()
+
+    def test_add_nancodes(self):
+        df = pd.DataFrame({
+            "timestamp": pd.date_range("20200321", "20200328"),
+            "geo_id": ["01017", "01043", "01061", "01103", "02282", "72001", "31000", "49000"],
+            "val": [0.1, 0.2, 0.3, 0.4, 0.5, np.nan, 0.7, np.nan],
+            "se": [np.nan] * 8,
+            "sample_size": [np.nan] * 8
+        }).set_index(["timestamp", "geo_id"])
+        expected_df = pd.DataFrame({
+            "timestamp": pd.date_range("20200321", "20200328"),
+            "geo_id": ["01017", "01043", "01061", "01103", "02282", "72001", "31000", "49000"],
+            "val": [0.1, 0.2, 0.3, 0.4, 0.5, np.nan, 0.7, np.nan],
+            "se": [np.nan] * 8,
+            "sample_size": [np.nan] * 8,
+            "missing_val": [Nans.NOT_MISSING] * 5 + [Nans.REGION_EXCEPTION, Nans.NOT_MISSING, Nans.UNKNOWN],
+            "missing_se": [Nans.NOT_APPLICABLE] * 8,
+            "missing_sample_size": [Nans.NOT_APPLICABLE] * 8,
+        }).set_index(["timestamp", "geo_id"])
+
+        pd.testing.assert_frame_equal(add_nancodes(df, "deaths", "county", None), expected_df)
+
+        df2 = pd.DataFrame({
+            "timestamp": pd.date_range("20200321", "20200328"),
+            "geo_id": ["01017", "01043", "01061", "01103", "02282", "72001", "31000", "49000"],
+            "val": [np.nan] * 6 + [0.7, np.nan],
+            "se": [np.nan] * 8,
+            "sample_size": [np.nan] * 8
+        }).set_index(["timestamp", "geo_id"])
+        expected_df2 = pd.DataFrame({
+            "timestamp": pd.date_range("20200321", "20200328"),
+            "geo_id": ["01017", "01043", "01061", "01103", "02282", "72001", "31000", "49000"],
+            "val": [np.nan] * 6 + [0.7, np.nan],
+            "se": [np.nan] * 8,
+            "sample_size": [np.nan] * 8,
+            "missing_val": [Nans.PRIVACY] * 5 + [Nans.REGION_EXCEPTION, Nans.NOT_MISSING, Nans.UNKNOWN],
+            "missing_se": [Nans.NOT_APPLICABLE] * 8,
+            "missing_sample_size": [Nans.NOT_APPLICABLE] * 8,
+        }).set_index(["timestamp", "geo_id"])
+
+        pd.testing.assert_frame_equal(add_nancodes(df2, "deaths", "county", "seven_day_average"), expected_df2)