Merge pull request #1478 from cmu-delphi/google_symptoms_omicron

New Google symptoms signals - omicron related

Author: korlaxxalrok

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"
       ]
     }
   },

google_symptoms/delphi_google_symptoms/constants.py

@@ -1,15 +1,33 @@
 """Registry for constants."""
-from functools import partial
 from datetime import timedelta
 
-from .smooth import (
-    identity,
-    kday_moving_average,
-)
+from delphi_utils import Smoother
 
 # global constants
-METRICS = ["Anosmia", "Ageusia"]
-COMBINED_METRIC = "sum_anosmia_ageusia"
+
+SYMPTOM_SETS = {
+    "s01": ["Cough", "Phlegm", "Sputum", "Upper respiratory tract infection"],
+    "s02": ["Nasal congestion", "Post nasal drip", "Rhinorrhea", "Sinusitis",
+            "Rhinitis", "Common cold"],
+    "s03": ["Fever", "Hyperthermia", "Chills", "Shivering", "Low grade fever"],
+    #"s04": ["Fatigue", "Weakness", "Muscle weakness", "Myalgia", "Pain"],
+    "s05": ["Shortness of breath", "Wheeze", "Croup", "Pneumonia", "Asthma",
+            "Crackles", "Acute bronchitis", "Bronchitis"],
+    "s06": ["Anosmia", "Dysgeusia", "Ageusia"],
+    #"s07": ["Nausea", "Vomiting", "Diarrhea", "Indigestion", "Abdominal pain"],
+    "s08": ["Laryngitis", "Sore throat", "Throat irritation"],
+    #"s09": ["Headache", "Migraine", "Cluster headache", "Dizziness", "Lightheadedness"],
+    #"s10": ["Night sweats","Perspiration", "hyperhidrosis"],
+    "scontrol": ["Type 2 diabetes", "Urinary tract infection", "Hair loss",
+                 "Candidiasis", "Weight gain"]
+}
+
+COMBINED_METRIC = list(SYMPTOM_SETS.keys())
+
+METRICS = list()
+for combmetric in COMBINED_METRIC:
+    METRICS = METRICS + SYMPTOM_SETS[combmetric]
+
 SMOOTHERS = ["raw", "smoothed"]
 GEO_RESOLUTIONS = [
     "state",
@@ -20,12 +38,15 @@
     "nation"
 ]
 
-seven_day_moving_average = partial(kday_moving_average, k=7)
 SMOOTHERS_MAP = {
-    "raw":           (identity, lambda d: d - timedelta(days=7)),
-    "smoothed":      (seven_day_moving_average, lambda d: d),
+    "raw":               (Smoother("identity", impute_method=None),
+                          lambda d: d - timedelta(days=7)),
+    "smoothed":          (Smoother("moving_average", window_length=7,
+                                   impute_method='zeros'), lambda d: d)
 }
 
+
+
 STATE_TO_ABBREV = {'Alabama': 'al',
                    'Alaska': 'ak',
                    #                   'American Samoa': 'as',

google_symptoms/delphi_google_symptoms/geo.py

@@ -48,7 +48,7 @@ def generate_transition_matrix(geo_res):
               ).fillna(0).reset_index().rename({mapping_flag: "geo_id"}, axis = 1)
     return map_df
 
-def geo_map(df, geo_res):
+def geo_map(df, geo_res, namescols =  None):
     """
     Compute derived HRR and MSA counts as a weighted sum of the county dataset.
 
@@ -59,6 +59,11 @@ def geo_map(df, geo_res):
         and columns for signal vals
     geo_res: str
         "msa", "hrr", "hhs" or "nation"
+    namescols: list of strings
+        names of columns of df but geo_id and timestamp
+        when running the pipeline, this will always be METRICS+COMBINED_METRIC
+        this parameter was added to allow us to run unit tests in subsets of
+        metrics and combined_metric's
 
     Returns
     -------
@@ -67,6 +72,9 @@ def geo_map(df, geo_res):
         and columns for signal vals.
         The geo_id has been converted from fips to HRRs/MSAs
     """
+    if namescols is None:
+        namescols = METRICS + COMBINED_METRIC
+
     if geo_res == "county":
         return df
 
@@ -75,7 +83,7 @@ def geo_map(df, geo_res):
     for _date in df["timestamp"].unique():
         val_lists = df[df["timestamp"] == _date].merge(
                 map_df["geo_id"], how="right"
-                )[METRICS + [COMBINED_METRIC]].fillna(0)
+                )[namescols].fillna(0)
         newdf = pd.DataFrame(
                 np.matmul(map_df.values[:, 1:].T, val_lists.values),
                 columns = list(val_lists.keys())

google_symptoms/delphi_google_symptoms/pull.py

@@ -7,7 +7,7 @@
 import numpy as np
 import pandas as pd
 
-from .constants import DC_FIPS, METRICS, COMBINED_METRIC
+from .constants import DC_FIPS, METRICS, COMBINED_METRIC, SYMPTOM_SETS
 
 
 # Create map of BigQuery symptom column names to desired column names.
@@ -39,18 +39,18 @@ def preprocess(df, level):
         Dataframe as described above.
     """
     # Constants
-    KEEP_COLUMNS = ["geo_id", "date"] + METRICS + [COMBINED_METRIC]
+    KEEP_COLUMNS = ["geo_id", "date"] + METRICS + COMBINED_METRIC
 
     df.rename(colname_map, axis=1, inplace=True)
     df["geo_id"] = df["open_covid_region_code"].apply(
         lambda x: x.split("-")[-1].lower())
 
-    df[COMBINED_METRIC] = 0
-    for metric in METRICS:
-        df[COMBINED_METRIC] += df[metric].fillna(0)
-    df.loc[
-        (df["Anosmia"].isnull())
-        & (df["Ageusia"].isnull()), COMBINED_METRIC] = np.nan
+    for cb_metric in COMBINED_METRIC:
+        df[cb_metric] = 0
+        for metric in SYMPTOM_SETS[cb_metric]:
+            df[cb_metric] += df[metric].fillna(0)
+        df[cb_metric] = df[cb_metric]/len(SYMPTOM_SETS[cb_metric])
+        df.loc[df[SYMPTOM_SETS[cb_metric]].isnull().all(axis=1), cb_metric] = np.nan
 
     # Delete rows with missing FIPS
     null_mask = (df["geo_id"].isnull())

google_symptoms/delphi_google_symptoms/run.py

@@ -17,7 +17,7 @@
 )
 from delphi_utils.validator.utils import lag_converter
 
-from .constants import (METRICS, COMBINED_METRIC,
+from .constants import (COMBINED_METRIC,
                         GEO_RESOLUTIONS, SMOOTHERS, SMOOTHERS_MAP)
 from .geo import geo_map
 from .pull import pull_gs_data
@@ -94,15 +94,16 @@ def run_module(params):
 
         if len(df_pull) == 0:
             continue
-        for metric, smoother in product(
-                METRICS+[COMBINED_METRIC], SMOOTHERS):
+        for metric, smoother in product(COMBINED_METRIC, SMOOTHERS):
             logger.info("generating signal and exporting to CSV",
                         geo_res=geo_res,
                         metric=metric,
                         smoother=smoother)
-            df = df_pull.set_index(["timestamp", "geo_id"])
-            df["val"] = df[metric].groupby(level=1
-                                           ).transform(SMOOTHERS_MAP[smoother][0])
+            df = df_pull
+            df["val"] = df[metric].astype(float)
+            df["val"] = df[["geo_id", "val"]].groupby(
+                "geo_id")["val"].transform(
+                SMOOTHERS_MAP[smoother][0].smooth)
             df["se"] = np.nan
             df["sample_size"] = np.nan
             # Drop early entries where data insufficient for smoothing

google_symptoms/delphi_google_symptoms/smooth.py

@@ -2,7 +2,7 @@
   "common": {
     "export_dir": "./receiving",
     "log_exceptions": false,
-    "log_filename": "./google-symptoms.log"   
+    "log_filename": "./google-symptoms.log"
   },
   "indicator": {
     "export_start_date": "2020-02-20",
@@ -26,9 +26,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"
       ]
     }
   },