Merge pull request #1377 from cmu-delphi/release/indicators_v0.2.7_utils_v0.2.5

Release covidcast-indicators 0.2.7

Author: krivard

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.2.6
+current_version = 0.2.7
 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.4
+current_version = 0.2.5
 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.4"
+__version__ = "0.2.5"

_delphi_utils_python/delphi_utils/archive.py

@@ -307,8 +307,9 @@ def filter_exports(self, common_diffs: FileDiffMap):
             else:
                 replace(diff_file, exported_file)
 
-    def run(self):
+    def run(self, logger=None):
         """Run the differ and archive the changed and new files."""
+        start_time = time.time()
         self.update_cache()
 
         # Diff exports, and make incremental versions
@@ -317,6 +318,11 @@ def run(self):
         # Archive changed and new files only
         to_archive = [f for f, diff in common_diffs.items()
                       if diff is not None]
+        if logger:
+            logger.debug("Diffed exports",
+                         phase="archiving",
+                         new_files_count=len(new_files),
+                         common_diffs_count=len(to_archive))
         to_archive += new_files
         _, fails = self.archive_exports(to_archive)
 
@@ -329,6 +335,14 @@ def run(self):
         for exported_file in fails:
             print(f"Failed to archive '{exported_file}'")
 
+        elapsed_time_in_seconds = round(time.time() - start_time, 2)
+        if logger:
+            logger.info("Completed archive run",
+                        phase="archiving",
+                        elapsed_time_in_seconds=elapsed_time_in_seconds,
+                        new_changed_count=len(to_archive),
+                        fail_count=len(fails))
+
 
 class S3ArchiveDiffer(ArchiveDiffer):
     """
@@ -641,8 +655,23 @@ def archive_exports(self, exported_files):
             successes: All files from input
             fails: Empty list
         """
+        archive_success, archive_fail = [], []
+        for exported_file in exported_files:
+            archive_file = abspath(
+                join(self.cache_dir, basename(exported_file)))
+
+            # Copy export to cache
+            try:
+                # Archive
+                shutil.copyfile(exported_file, archive_file)
+                archive_success.append(exported_file)
+
+            except FileNotFoundError as ex:
+                print(ex)
+                archive_fail.append(exported_file)
+
         self._exports_archived = True
-        return exported_files, []
+        return archive_success, archive_fail
 
     def update_cache(self):
         """Handle cache updates with a no-op.
@@ -667,13 +696,6 @@ def update_cache(self):
     if "archive" not in _params:
         _params = {"archive": _params, "common": _params}
 
-    logger = get_structured_logger(
+    archiver_from_params(_params).run(get_structured_logger(
         __name__, filename=_params["common"].get("log_filename"),
-        log_exceptions=_params["common"].get("log_exceptions", True))
-    start_time = time.time()
-
-    archiver_from_params(_params).run()
-
-    elapsed_time_in_seconds = round(time.time() - start_time, 2)
-    logger.info("Completed archive run.",
-                elapsed_time_in_seconds=elapsed_time_in_seconds)
+        log_exceptions=_params["common"].get("log_exceptions", True)))

_delphi_utils_python/delphi_utils/runner.py

@@ -36,17 +36,19 @@ def run_indicator_pipeline(indicator_fn:  Callable[[Params], None],
         None if no archiving should be performed.
     """
     params = read_params()
+    logger = get_structured_logger(
+        name=indicator_fn.__module__,
+        filename=params["common"].get("log_filename", None),
+        log_exceptions=params["common"].get("log_exceptions", True))
     indicator_fn(params)
     validator = validator_fn(params)
     archiver = archiver_fn(params)
     if validator:
         validation_report = validator.validate()
-        validation_report.log(get_structured_logger(
-            name = indicator_fn.__module__,
-            filename=params["common"].get("log_filename", None)))
+        validation_report.log(logger)
     if (not validator or validation_report.success()):
         if archiver:
-            archiver.run()
+            archiver.run(logger)
         if "delivery" in params:
             transfer_files()
 

_delphi_utils_python/setup.py

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

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

@@ -66,6 +66,7 @@
         "smoothed_vaccine_likely_doctors", "smoothed_wvaccine_likely_doctors",
         "smoothed_felt_isolated_7d", "smoothed_wfelt_isolated_7d",
         "smoothed_worried_become_ill", "smoothed_wworried_become_ill",
+        "smoothed_received_2_vaccine_doses", "smoothed_wreceived_2_vaccine_doses",
         ["smoothed_vaccine_barrier_appointment_time_tried", "msa"], ["smoothed_wvaccine_barrier_appointment_time_tried", "msa"],
         ["smoothed_vaccine_barrier_childcare_tried", "msa"], ["smoothed_wvaccine_barrier_childcare_tried", "msa"],
         ["smoothed_vaccine_barrier_document_tried", "msa"], ["smoothed_wvaccine_barrier_document_tried", "msa"],
@@ -103,7 +104,7 @@
       "maintainers": []
     },
     "covid-act-now": {
-      "max_age":5,
+      "max_age":9,
       "maintainers": []
     },
     "hhs": {

facebook/delphiFacebook/R/contingency_aggregate.R

@@ -158,15 +158,20 @@ post_process_aggs <- function(df, aggregations, cw_list) {
   metric_cols <- unique(aggregations$metric)
 
   cols_check_available <- unique(c(group_vars[group_vars != "geo_id"], metric_cols))
-  available <- cols_check_available %in% names(df)
+  available <- unlist(
+    lapply(cols_check_available, function(col) {
+      # Exists in dataframe and column is not only NAs
+      col %in% names(df) && !all(is.na( unique(df[[col]] )))
+    })
+  )
   cols_not_available <- cols_check_available[ !available ]
   for (col_var in cols_not_available) {
     # Remove from aggregations
     aggregations <- aggregations[aggregations$metric != col_var &
                                    !mapply(aggregations$group_by,
                                            FUN=function(x) {col_var %in% x}), ]
     msg_plain(paste0(
-        col_var, " is not defined. Removing all aggregations that use it. ",
+        col_var, " is not available or not defined. Removing all aggregations that use it. ",
         nrow(aggregations), " remaining")
     )
   }

hhs_hosp/delphi_hhs/run.py

@@ -117,6 +117,11 @@ def run_module(params):
             df=pop_proportion(df, geo_mapper)
         df = make_geo(df, geo, geo_mapper)
         df = smooth_values(df, smoother[0])
+        # Fix N/A MA values, see issue #1360
+        if geo == "state" and sensor.startswith(CONFIRMED_FLU):
+            ma_filter = df.val.isna() & (df.geo_id == "ma") & (df.timestamp > "08-01-2021") & \
+                (df.timestamp.dt.day_name() == "Tuesday")
+            df = df[~ma_filter]
         if df.empty:
             continue
         sensor_name = sensor + smoother[1]

sir_complainsalot/params.json.template

@@ -66,6 +66,7 @@
         "smoothed_vaccine_likely_doctors", "smoothed_wvaccine_likely_doctors",
         "smoothed_felt_isolated_7d", "smoothed_wfelt_isolated_7d",
         "smoothed_worried_become_ill", "smoothed_wworried_become_ill",
+        "smoothed_received_2_vaccine_doses", "smoothed_wreceived_2_vaccine_doses",
         ["smoothed_vaccine_barrier_appointment_time_tried", "msa"], ["smoothed_wvaccine_barrier_appointment_time_tried", "msa"],
         ["smoothed_vaccine_barrier_childcare_tried", "msa"], ["smoothed_wvaccine_barrier_childcare_tried", "msa"],
         ["smoothed_vaccine_barrier_document_tried", "msa"], ["smoothed_wvaccine_barrier_document_tried", "msa"],
@@ -102,7 +103,7 @@
       "maintainers": []
     },
     "covid-act-now": {
-      "max_age":5,
+      "max_age":9,
       "maintainers": []
     },
     "hhs": {