Merge pull request #1617 from cmu-delphi/release/indicators_v0.3.13_utils_v0.3.4

Release covidcast-indicators 0.3.13

Author: krivard

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.3.12
+current_version = 0.3.13
 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.3.3
+current_version = 0.3.4
 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.3.3"
+__version__ = "0.3.4"

_delphi_utils_python/delphi_utils/validator/dynamic.py

@@ -273,8 +273,8 @@ def create_dfs(self, geo_sig_df, api_df_or_error, checking_date, geo_type, signa
             - report: ValidationReport; report where results are added
 
         Returns:
-            - False if recent_df is empty, else (recent_df, reference_api_df)
-            (after reference_api_df has been padded if necessary)
+            - False if recent_df is empty after padding, else (recent_df, reference_api_df)
+            (reference_api_df will be padded if necessary)
         """
         # recent_lookbehind: start from the check date and working backward in time,
         # how many days at a time do we want to check for anomalies?
@@ -320,47 +320,71 @@ def create_dfs(self, geo_sig_df, api_df_or_error, checking_date, geo_type, signa
         reference_api_df = api_df_or_error.query(
             "time_value >= @reference_start_date & time_value <= @reference_end_date")
 
-        report.increment_total_checks()
+        pre_pad_empty_flag = reference_api_df.empty
+        reference_api_df = self.pad_reference_api_df(
+            reference_api_df, geo_sig_df, reference_start_date, reference_end_date)
 
+        report.increment_total_checks()
         if reference_api_df.empty:
-            report.add_raised_error(
-                ValidationFailure("empty_reference_data",
+            report.add_raised_error(ValidationFailure("empty_reference_data",
                                   checking_date,
                                   geo_type,
                                   signal_type,
                                   "reference data is empty; comparative checks could not "
                                   "be performed"))
             return False
-
-        reference_api_df = self.pad_reference_api_df(
-            reference_api_df, geo_sig_df, reference_end_date)
+        if pre_pad_empty_flag:
+            report.add_raised_warning(ValidationFailure("empty_reference_data",
+                                  checking_date,
+                                  geo_type,
+                                  signal_type,
+                                  "pre-padding reference data is empty and indicates data "
+                                  "missing from the API; please verify that this is expected"))
 
         return (geo_sig_df, reference_api_df)
 
-    def pad_reference_api_df(self, reference_api_df, geo_sig_df, reference_end_date):
+    # `reference_start_date` is used in the call to `geo_sig_df.query()`
+    # below but pylint doesn't recognize that.
+    # pylint: disable=unused-argument
+    def pad_reference_api_df(self, reference_api_df, geo_sig_df,
+                             reference_start_date, reference_end_date):
         """Check if API data is missing, and supplement from test data.
 
         Arguments:
             - reference_api_df: API data within lookbehind range
             - geo_sig_df: Test data
-            - reference_end_date: Supposed end date of reference data
+            - reference_start_date: Desired start date of reference data
+            - reference_end_date: Desired end date of reference data
 
         Returns:
             - reference_api_df: Supplemented version of original
         """
+        # Value is `NaT` (not a time) if reference_api_df is empty.
         reference_api_df_max_date = reference_api_df.time_value.max().date()
-        if reference_api_df_max_date < reference_end_date:
-            # Querying geo_sig_df, only taking relevant rows
+        if reference_api_df.empty:
+            geo_sig_df_supplement = geo_sig_df.query(
+                'time_value <= @reference_end_date & time_value >= \
+                @reference_start_date')[[
+                "geo_id", "val", "se", "sample_size", "time_value"]]
+        elif reference_api_df_max_date < reference_end_date:
+            # If actual end date `reference_api_df_max_date` is not as recent as
+            # the desired end date `reference_end_date`, add rows from recently
+            # generate data, in `geo_sig_df`, to the reference data.
             geo_sig_df_supplement = geo_sig_df.query(
                 'time_value <= @reference_end_date & time_value > \
                 @reference_api_df_max_date')[[
                 "geo_id", "val", "se", "sample_size", "time_value"]]
-            # Matching time_value format
-            geo_sig_df_supplement["time_value"] = \
-                pd.to_datetime(geo_sig_df_supplement["time_value"],
-                    format = "%Y-%m-%d %H:%M:%S")
-            reference_api_df = pd.concat(
-                [reference_api_df, geo_sig_df_supplement])
+        else:
+            return reference_api_df
+
+        # Final processing after supplementing reference_api_df
+        # Matching time_value format
+        geo_sig_df_supplement["time_value"] = \
+            pd.to_datetime(geo_sig_df_supplement["time_value"],
+                format = "%Y-%m-%d %H:%M:%S")
+        reference_api_df = pd.concat(
+            [reference_api_df, geo_sig_df_supplement])
+
         return reference_api_df
 
     def check_max_date_vs_reference(self, df_to_test, df_to_reference, checking_date,

_delphi_utils_python/setup.py

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

_delphi_utils_python/tests/validator/test_dynamic.py

@@ -25,8 +25,9 @@ def test_no_padding(self):
         test_df = pd.DataFrame(data)
         ref_df = pd.DataFrame(data)
 
+        ref_date = datetime.strptime("2021-01-06", "%Y-%m-%d").date()
         new_ref_df = validator.pad_reference_api_df(
-            ref_df, test_df, datetime.strptime("2021-01-06", "%Y-%m-%d").date())
+            ref_df, test_df, ref_date, ref_date)
 
         assert new_ref_df.equals(ref_df)
 
@@ -42,8 +43,9 @@ def test_half_padding(self):
         ref_df = pd.DataFrame(ref_data)
         test_df = pd.DataFrame(test_data)
 
+        ref_date = datetime.strptime("2021-01-15", "%Y-%m-%d").date()
         new_ref_df = validator.pad_reference_api_df(
-            ref_df, test_df, datetime.strptime("2021-01-15", "%Y-%m-%d").date())
+            ref_df, test_df, ref_date, ref_date)
 
         # Check it only takes missing dates - so the last 5 dates
         assert new_ref_df.time_value.max() == datetime.strptime("2021-01-11",
@@ -64,8 +66,9 @@ def test_full_padding(self):
         ref_df = pd.DataFrame(ref_data)
         test_df = pd.DataFrame(test_data)
 
+        ref_date = datetime.strptime("2021-01-15", "%Y-%m-%d").date()
         new_ref_df = validator.pad_reference_api_df(
-            ref_df, test_df, datetime.strptime("2021-01-15", "%Y-%m-%d").date())
+            ref_df, test_df, ref_date, ref_date)
 
         # Check it only takes missing dates up to the day before the reference
         assert new_ref_df.time_value.max() == datetime.strptime("2021-01-15",

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

@@ -11,7 +11,8 @@
       "doses administered",
       "booster doses administered",
       "fully vaccinated",
-      "booster dose since"
+      "booster dose since",
+      "positivity"
     ]
   },
   "validation": {
@@ -31,7 +32,6 @@
     "dynamic": {
       "ref_window_size": 7,
       "smoothed_signals": [
-        "naats_total_7dav",
         "naats_positivity_7dav",
         "confirmed_admissions_covid_1d_prop_7dav",
         "confirmed_admissions_covid_1d_7dav",

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

@@ -133,8 +133,8 @@
         "smoothed_vaccine_barrier_none_has", "smoothed_wvaccine_barrier_none_has",
         "smoothed_vaccine_barrier_appointment_location_has", "smoothed_wvaccine_barrier_appointment_location_has",
         "smoothed_vaccine_barrier_other_has", "smoothed_wvaccine_barrier_other_has",
-        ["smoothed_vaccine_barrier_appointment_location_tried", "county", "state"],
-        ["smoothed_vaccine_barrier_other_tried", "county", "state"]
+        ["smoothed_vaccine_barrier_appointment_location_tried", "county", "state"], ["smoothed_wvaccine_barrier_appointment_location_tried", "county", "state"],
+        ["smoothed_vaccine_barrier_other_tried", "county", "state"], ["smoothed_wvaccine_barrier_other_tried", "county", "state"]
       ]
     },
     "quidel": {

dsew_community_profile/delphi_dsew_community_profile/pull.py

@@ -407,8 +407,11 @@ def as_cached_filename(params, config):
 
 def fetch_listing(params):
     """Generate the list of report files to process."""
-    listing = requests.get(DOWNLOAD_LISTING).json()['metadata']['attachments']
+    export_start_date = params['indicator'].get(
+        'export_start_date', datetime.datetime.fromtimestamp(0).date()
+    )
 
+    listing = requests.get(DOWNLOAD_LISTING).json()['metadata']['attachments']
     # drop the pdf files
     listing = [
         dict(
@@ -418,33 +421,34 @@ def fetch_listing(params):
         )
         for el in listing if el['filename'].endswith("xlsx")
     ]
-    keep = []
+
+    def check_valid_publish_date(x):
+        return x['publish_date'] >= export_start_date
+
     if params['indicator']['reports'] == 'new':
         # drop files we already have in the input cache
-        keep = [el for el in listing if not os.path.exists(el['cached_filename'])]
+        keep = [
+            el for el in listing
+            if not os.path.exists(el['cached_filename']) and check_valid_publish_date(el)
+        ]
     elif params['indicator']['reports'].find("--") > 0:
         # drop files outside the specified publish-date range
         start_str, _, end_str = params['indicator']['reports'].partition("--")
         start_date = datetime.datetime.strptime(start_str, "%Y-%m-%d").date()
         end_date = datetime.datetime.strptime(end_str, "%Y-%m-%d").date()
         keep = [
             el for el in listing
-            if start_date <= el['publish_date'] <= end_date
+            if (start_date <= el['publish_date'] <= end_date) and check_valid_publish_date(el)
         ]
-
-    # reference date is guaranteed to be on or before publish date, so we can trim
-    # reports that are too early
-    if 'export_start_date' in params['indicator']:
+    elif params['indicator']['reports'] == 'all':
         keep = [
-            el for el in keep
-            if params['indicator']['export_start_date'] <= el['publish_date']
+            el for el in listing if check_valid_publish_date(el)
         ]
-    # can't do the same for export_end_date
+    else:
+        raise ValueError("params['indicator']['reports'] is set to" \
+            + f" {params['indicator']['reports']}, which isn't 'new', 'all', or a date range.")
 
-    # if we're only running on a subset, make sure we have enough data for interp
-    if keep:
-        keep = extend_listing_for_interp(keep, listing)
-    return keep if keep else listing
+    return extend_listing_for_interp(keep, listing)
 
 def extend_listing_for_interp(keep, listing):
     """Grab additional files from the full listing for interpolation if needed.
@@ -617,7 +621,8 @@ def interpolate_missing_values(dfs: DataDict) -> DataDict:
         # https://github.com/cmu-delphi/covidcast-indicators/issues/1576
         _, sig, _ = key
         if sig == "positivity":
-            interpolate_df[key] = df.set_index(["geo_id", "timestamp"]).sort_index().reset_index()
+            reindexed_group_df = df.set_index(["geo_id", "timestamp"]).sort_index().reset_index()
+            interpolate_df[key] = reindexed_group_df[~reindexed_group_df.val.isna()]
             continue
 
         geo_dfs = []

dsew_community_profile/delphi_dsew_community_profile/run.py

@@ -99,20 +99,20 @@ def replace_date_param(p):
             params['indicator']['reports'] = 'all'
             params['indicator']['export_signals'] = {sensor_names[key] for key in new_signals}
 
-        dfs = fetch_new_reports(params, logger)
-        for key, df in dfs.items():
-            (geo, sig, is_prop) = key
-            if sig not in params["indicator"]["export_signals"]:
-                continue
-            dates = create_export_csv(
-                df,
-                params['common']['export_dir'],
-                geo,
-                make_signal_name(sig, is_prop),
-                **export_params
-            )
-            if len(dates)>0:
-                run_stats.append((max(dates), len(dates)))
+            dfs = fetch_new_reports(params, logger)
+            for key, df in dfs.items():
+                (geo, sig, is_prop) = key
+                if sig not in params["indicator"]["export_signals"]:
+                    continue
+                dates = create_export_csv(
+                    df,
+                    params['common']['export_dir'],
+                    geo,
+                    make_signal_name(sig, is_prop),
+                    **export_params
+                )
+                if len(dates)>0:
+                    run_stats.append((max(dates), len(dates)))
 
     ## log this indicator run
     elapsed_time_in_seconds = round(time.time() - start_time, 2)

dsew_community_profile/params.json.template

@@ -10,7 +10,6 @@
     "export_end_date": null,
     "export_signals": [
       "confirmed covid-19 admissions",
-      "total",
       "positivity",
       "doses administered",
       "booster doses administered",
@@ -35,7 +34,6 @@
     "dynamic": {
       "ref_window_size": 7,
       "smoothed_signals": [
-        "naats_total_7dav",
         "naats_positivity_7dav",
         "confirmed_admissions_covid_1d_prop_7dav",
         "confirmed_admissions_covid_1d_7dav",

dsew_community_profile/tests/test_pull.py

@@ -569,6 +569,11 @@ def test_extend_listing(self):
                 [{"publish_date": date(2020, 1, 20)}, {"publish_date": date(2020, 1, 19)}],
                 [{"publish_date": date(2020, 1, 20)}, {"publish_date": date(2020, 1, 19)}, {"publish_date": date(2020, 1, 18)}]
             ),
+            # empty keep list
+            example(
+                [],
+                []
+            )
         ]
         for ex in examples:
             assert extend_listing_for_interp(ex.given, listing) == ex.expected, ex.given

facebook/qsf-tools/static/CMU/item_replacement_map.csv

@@ -1,4 +1,5 @@
 new_item,old_item
+A5,A2b
 B8,B5
 B9,B5
 B10,B5
@@ -24,6 +25,7 @@ C13c,C13a
 C14a,C14
 C17,C2
 C17a,C17
+C17b,C17a
 V2a,V2
 V3a,V3
 V4a_1,V4_1

sir_complainsalot/params.json.template

@@ -133,8 +133,8 @@
         "smoothed_vaccine_barrier_none_has", "smoothed_wvaccine_barrier_none_has",
         "smoothed_vaccine_barrier_appointment_location_has", "smoothed_wvaccine_barrier_appointment_location_has",
         "smoothed_vaccine_barrier_other_has", "smoothed_wvaccine_barrier_other_has",
-        ["smoothed_vaccine_barrier_appointment_location_tried", "county", "state"],
-        ["smoothed_vaccine_barrier_other_tried", "county", "state"]
+        ["smoothed_vaccine_barrier_appointment_location_tried", "county", "state"], ["smoothed_wvaccine_barrier_appointment_location_tried", "county", "state"],
+        ["smoothed_vaccine_barrier_other_tried", "county", "state"], ["smoothed_wvaccine_barrier_other_tried", "county", "state"]
       ]
     },
     "quidel": {