feat: add scoring to covid prod

Author: dsweber2

R/aux_data_utils.R

@@ -719,10 +719,12 @@ create_nhsn_data_archive <- function(disease_name) {
     as_epi_archive(compactify = TRUE)
 }
 
-
 up_to_date_nssp_state_archive <- function(disease = c("covid", "influenza")) {
   disease <- arg_match(disease)
-  nssp_state <- pub_covidcast(
+  nssp_state <- retry_fn(
+    max_attempts = 10,
+    wait_seconds = 1,
+    fn = pub_covidcast,
     source = "nssp",
     signal = glue::glue("pct_ed_visits_{disease}"),
     time_type = "week",

R/utils.R

@@ -162,7 +162,7 @@ data_substitutions <- function(dataset, disease, forecast_generation_date) {
 }
 
 parse_prod_weights <- function(filename = here::here("covid_geo_exclusions.csv"),
-                               forecast_date_int, forecaster_fns) {
+                               forecast_date_int, forecaster_fn_names) {
   forecast_date <- as.Date(forecast_date_int)
   all_states <- c(
     unique(readr::read_csv("https://raw.githubusercontent.com/cmu-delphi/covidcast-indicators/refs/heads/main/_delphi_utils_python/delphi_utils/data/2020/state_pop.csv", show_col_types = FALSE)$state_id),
@@ -182,7 +182,7 @@ parse_prod_weights <- function(filename = here::here("covid_geo_exclusions.csv")
     ) %>%
     unnest_longer(geo_value) %>%
     mutate(
-      forecaster = ifelse(forecaster == "all", list(names(forecaster_fns)), forecaster),
+      forecaster = ifelse(forecaster == "all", list(forecaster_fn_names), forecaster),
     ) %>%
     unnest_longer(forecaster) %>%
     group_by(forecast_date, forecaster, geo_value) %>%
@@ -472,3 +472,39 @@ get_recent_targets_errors <- function(time_since = minutes(60)) {
 
   return(invisible(meta_df %>% filter(time > Sys.time() - time_since)))
 }
+
+#' Retry a function.
+#'
+#' @param max_attempts The maximum number of attempts.
+#' @param wait_seconds The number of seconds to wait between attempts.
+#' @param fn The function to retry.
+#' @param ... Additional arguments to pass to the function.
+#'
+#' @examples
+#' retry_fn(
+#'   max_attempts = 10,
+#'   wait_seconds = 1,
+#'   fn = pub_covidcast,
+#'   source = "nssp",
+#'   signal = "pct_ed_visits_covid",
+#'   geo_type = "state",
+#'   geo_values = "*",
+#'   time_type = "week"
+#' )
+retry_fn <- function(max_attempts = 10, wait_seconds = 1, fn, ...) {
+  for (attempt in 1:max_attempts) {
+    tryCatch(
+      {
+        result <- fn(...)
+        return(result) # Return successful result
+      },
+      error = function(e) {
+        if (attempt == max_attempts) {
+          stop("Maximum retry attempts reached. Last error: ", e$message)
+        }
+        message(sprintf("Attempt %d failed. Retrying in %d second(s)...", attempt, wait_seconds))
+        Sys.sleep(wait_seconds)
+      }
+    )
+  }
+}

_targets.yaml

@@ -2,11 +2,12 @@ covid_hosp_explore:
   script: scripts/covid_hosp_explore.R
   store: covid_hosp_explore
   use_crew: no
-  error: null
+  reporter_make: timestamp
 flu_hosp_explore:
   script: scripts/flu_hosp_explore.R
   store: flu_hosp_explore
   use_crew: no
+  reporter_make: timestamp
 flu_hosp_tiny:
   script: scripts/flu_hosp_tiny.R
   store: flu_hosp_tiny
@@ -15,7 +16,9 @@ flu_hosp_prod:
   script: scripts/flu_hosp_prod.R
   store: flu_hosp_prod
   use_crew: no
+  reporter_make: timestamp
 covid_hosp_prod:
   script: scripts/covid_hosp_prod.R
   store: covid_hosp_prod
   use_crew: no
+  reporter_make: timestamp

scripts/covid_hosp_explore.R

@@ -235,7 +235,10 @@ data_targets <- list2(
   tar_target(
     name = hhs_evaluation_data,
     command = {
-      epidatr::pub_covidcast(
+      retry_fn(
+        max_attempts = 10,
+        wait_seconds = 1,
+        fn = pub_covidcast,
         source = "hhs",
         signals = hhs_signal,
         geo_type = "state",
@@ -262,15 +265,21 @@ data_targets <- list2(
   tar_target(
     name = nssp_archive,
     command = {
-      nssp_state <- pub_covidcast(
+      nssp_state <- retry_fn(
+        max_attempts = 10,
+        wait_seconds = 1,
+        fn = pub_covidcast,
         source = "nssp",
         signal = "pct_ed_visits_covid",
         time_type = "week",
         geo_type = "state",
         geo_values = "*",
         fetch_args = epidatr::fetch_args_list(timeout_seconds = 400)
       )
-      nssp_hhs <- pub_covidcast(
+      nssp_hhs <- retry_fn(
+        max_attempts = 10,
+        wait_seconds = 1,
+        fn = pub_covidcast,
         source = "nssp",
         signal = "pct_ed_visits_covid",
         time_type = "week",
@@ -297,14 +306,20 @@ data_targets <- list2(
       # source going down completely, which means we're actually just comparing
       # with the version without this source
       all_of_them <- lapply(used_searches, \(search_name) {
-        google_symptoms_state_archive <- pub_covidcast(
+        google_symptoms_state_archive <- retry_fn(
+          max_attempts = 10,
+          wait_seconds = 1,
+          fn = pub_covidcast,
           source = "google-symptoms",
           signal = glue::glue("s0{search_name}_smoothed_search"),
           time_type = "day",
           geo_type = "state",
           geo_values = "*"
         )
-        google_symptoms_hhs_archive <- pub_covidcast(
+        google_symptoms_hhs_archive <- retry_fn(
+          max_attempts = 10,
+          wait_seconds = 1,
+          fn = pub_covidcast,
           source = "google-symptoms",
           signal = glue::glue("s0{search_name}_smoothed_search"),
           time_type = "day",