Merge pull request #1486 from cmu-delphi/release/indicators_v0.2.23_utils_v0.2.11

Release covidcast-indicators 0.2.23

Author: korlaxxalrok

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.2.22
+current_version = 0.2.23
 commit = True
 message = chore: bump covidcast-indicators to {new_version}
 tag = False

_delphi_utils_python/.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.2.10
+current_version = 0.2.11
 commit = True
 message = chore: bump delphi_utils to {new_version}
 tag = False

_delphi_utils_python/delphi_utils/__init__.py

@@ -15,4 +15,4 @@
 from .nancodes import Nans
 from .weekday import Weekday
 
-__version__ = "0.2.10"
+__version__ = "0.2.11"

_delphi_utils_python/setup.py

@@ -26,7 +26,7 @@
 
 setup(
     name="delphi_utils",
-    version="0.2.10",
+    version="0.2.11",
     description="Shared Utility Functions for Indicators",
     long_description=long_description,
     long_description_content_type="text/markdown",

_delphi_utils_python/tests/test_geomap.py

@@ -104,6 +104,14 @@ class TestGeoMapper:
             ),
         )
     )
+    mega_data_3 = pd.DataFrame(
+        {
+            "fips": [1123, 1125, 1126, 1128, 1129, 18181],
+            "timestamp": [pd.Timestamp("2018-01-01")] * 6,
+            "visits": [4, 1, 2, 5, 10, 100001],
+            "count": [2, 1, 5, 7, 3, 10021],
+        }
+    )
     jhu_uid_data = pd.DataFrame(
         {
             "jhu_uid": [
@@ -208,6 +216,27 @@ def test_megacounty(self, geomapper):
             new_data[["count"]].sum() - self.mega_data[["count"]].sum()
         ).sum() < 1e-3
 
+        new_data = geomapper.fips_to_megacounty(self.mega_data_3, 4, 1)
+        expected_df = pd.DataFrame(
+            {
+                "megafips": ["01000", "01128", "01129", "18181"],
+                "timestamp": [pd.Timestamp("2018-01-01")] * 4,
+                "visits": [7, 5, 10, 100001],
+                "count": [8, 7, 3, 10021],
+            }
+        )
+        pd.testing.assert_frame_equal(new_data.set_index("megafips").sort_index(axis=1), expected_df.set_index("megafips").sort_index(axis=1))
+        new_data = geomapper.fips_to_megacounty(self.mega_data_3, 4, 1, thr_col="count")
+        expected_df = pd.DataFrame(
+            {
+                "megafips": ["01000", "01126", "01128", "18181"],
+                "timestamp": [pd.Timestamp("2018-01-01")] * 4,
+                "visits": [15, 2, 5, 100001],
+                "count": [6, 5, 7, 10021],
+            }
+        )
+        pd.testing.assert_frame_equal(new_data.set_index("megafips").sort_index(axis=1), expected_df.set_index("megafips").sort_index(axis=1))
+
     def test_add_population_column(self, geomapper):
         new_data = geomapper.add_population_column(self.fips_data_3, "fips")
         assert new_data.shape == (5, 5)

ansible/templates/dsew_community_profile-params-prod.json.j2

@@ -5,7 +5,8 @@
   },
   "indicator": {
     "input_cache": "./input_cache",
-    "reports": "new"
+    "reports": "new",
+    "export_signals": ["confirmed covid-19 admissions"]
   },
   "validation": {
     "common": {

ansible/templates/google_symptoms-params-prod.json.j2

@@ -36,9 +36,13 @@
     "dynamic": {
       "ref_window_size": 7,
       "smoothed_signals": [
-        "ageusia_smoothed_search",
-        "sum_anosmia_ageusia_smoothed_search",
-        "anosmia_smoothed_search"
+        "s01_smoothed_search",
+        "s02_smoothed_search",
+        "s03_smoothed_search",
+        "s05_smoothed_search",
+        "s06_smoothed_search",
+        "s08_smoothed_search",
+        "scontrol_smoothed_search"
       ]
     }
   },

dsew_community_profile/README.md

@@ -23,6 +23,8 @@ Indicator-specific parameters:
   and exported.
 * `export_start_date`: a YYYY-mm-dd string indicating the first date to export.
 * `export_end_date`: a YYYY-mm-dd string indicating the final date to export.
+* `export_signals`: list of string keys from constants.SIGNALS indicating which
+  signals to export
 
 ## Running the Indicator
 

dsew_community_profile/delphi_dsew_community_profile/pull.py

@@ -342,12 +342,16 @@ def nation_from_state(df, sig, geomapper):
         ).drop(
             "norm_denom", axis=1
         )
-    return geomapper.replace_geocode(
+    df = geomapper.replace_geocode(
         df,
         'state_id',
         'nation',
         new_col="geo_id"
     )
+    df["se"] = None
+    df["sample_size"] = None
+
+    return df
 
 def fetch_new_reports(params, logger=None):
     """Retrieve, compute, and collate all data we haven't seen yet."""
@@ -367,10 +371,21 @@ def fetch_new_reports(params, logger=None):
             lambda x: x[x["publish_date"] == x["publish_date"].max()]
         ).drop(
             "publish_date", axis=1
+        ).drop_duplicates(
         )
 
         if len(latest_sig_df.index) > 0:
-            ret[sig] = latest_sig_df.reset_index(drop=True)
+            latest_sig_df = latest_sig_df.reset_index(drop=True)
+
+            assert all(latest_sig_df.groupby(
+                    ["timestamp", "geo_id"]
+                ).size(
+                ).reset_index(
+                    drop=True
+                ) == 1), f"Duplicate rows in {sig} indicate that one or" \
+                + " more reports was published multiple times and the copies differ"
+
+            ret[sig] = latest_sig_df
 
     # add nation from state
     geomapper = GeoMapper()

dsew_community_profile/delphi_dsew_community_profile/run.py

@@ -38,9 +38,12 @@ def run_module(params):
         __name__, filename=params["common"].get("log_filename"),
         log_exceptions=params["common"].get("log_exceptions", True))
     def replace_date_param(p):
-        if p in params["indicator"] and params["indicator"][p] is not None:
-            date_param = datetime.strptime(params["indicator"][p], "%Y-%m-%d").date()
-            params["indicator"][p] = date_param
+        if p in params["indicator"]:
+            if params["indicator"][p] is None:
+                del params["indicator"][p]
+            else:
+                date_param = datetime.strptime(params["indicator"][p], "%Y-%m-%d").date()
+                params["indicator"][p] = date_param
     replace_date_param("export_start_date")
     replace_date_param("export_end_date")
     export_params = {
@@ -56,6 +59,8 @@ def replace_date_param(p):
     dfs = fetch_new_reports(params, logger)
     for key, df in dfs.items():
         (geo, sig) = key
+        if sig not in params["indicator"]["export_signals"]:
+            continue
         dates = create_export_csv(
             df,
             params['common']['export_dir'],