Merge pull request #83 from cmu-delphi/aws_fix

Aws fix

Author: dsweber2

Makefile

@@ -7,3 +7,9 @@ run:
 
 run-nohup:
 	nohup Rscript run.R &
+
+sync:
+	Rscript sync.R
+
+dashboard:
+	Rscript dashboard.R

NAMESPACE

@@ -27,6 +27,7 @@ export(make_shared_ensembles)
 export(make_shared_grids)
 export(make_target_ensemble_grid)
 export(make_target_param_grid)
+export(manage_S3_forecast_cache)
 export(overprediction)
 export(perform_sanity_checks)
 export(read_external_predictions_data)
@@ -39,6 +40,8 @@ export(slide_forecaster)
 export(underprediction)
 export(weighted_interval_score)
 importFrom(assertthat,assert_that)
+importFrom(aws.s3,get_bucket)
+importFrom(aws.s3,s3sync)
 importFrom(cli,cli_abort)
 importFrom(cli,hash_animal)
 importFrom(dplyr,across)
@@ -82,6 +85,7 @@ importFrom(epipredict,step_population_scaling)
 importFrom(epipredict,step_training_window)
 importFrom(epiprocess,as_epi_df)
 importFrom(epiprocess,epix_slide)
+importFrom(here,here)
 importFrom(magrittr,"%<>%")
 importFrom(magrittr,"%>%")
 importFrom(purrr,imap)

R/manage_S3.R

@@ -0,0 +1,47 @@
+#' Manage S3 cache
+#' @param rel_cache_dir The relative path to the cache directory, e.g.
+#' "data-processed/2021-09-01". Default is `"{tar_project}/objects"`
+#' @param bucket_name The name of the S3 bucket to sync.
+#' @param direction Set 'download' to download files, 'upload' to upload files,
+#'   and `sync` to do both
+#' @param verbose Set to TRUE to print the files being synced.
+#' @param prefix specify the prefix for `s3sync`, which filters down which files
+#'   to sync to those starting with `prefix`.
+#' @param tar_project which targets project we're working on
+#' @importFrom aws.s3 s3sync get_bucket
+#' @importFrom here here
+#' @export
+manage_S3_forecast_cache <- function(rel_cache_dir = NULL,
+                                     bucket_name = "forecasting-team-data",
+                                     direction = "sync",
+                                     verbose = FALSE,
+                                     prefix = Sys.getenv("AWS_S3_PREFIX", "exploration"),
+                                     tar_project = Sys.getenv("TAR_PROJECT", "exploration")) {
+  if (is.null(rel_cache_dir)) {
+    cache_path <- tar_project
+  } else {
+    cache_path <- here(rel_cache_dir)
+  }
+  if (!dir.exists(cache_path)) dir.create(cache_path)
+
+  prefix <- paste0(prefix, "/", tar_project, "/")
+  s3b <- get_bucket(bucket_name, prefix = prefix)
+  if (direction == "sync") {
+    if (verbose) {
+      s3sync(cache_path, s3b, prefix = prefix)
+    } else {
+      sink("/dev/null")
+      s3sync(cache_path, s3b, prefix = prefix, verbose = FALSE)
+      sink()
+    }
+  } else {
+    if (verbose) {
+      s3sync(cache_path, s3b, prefix = prefix, direction = direction)
+    } else {
+      sink("/dev/null")
+      s3sync(cache_path, s3b, prefix = prefix, direction = direction, verbose = FALSE)
+      sink()
+    }
+  }
+  return(TRUE)
+}

R/targets_utils.R

@@ -273,7 +273,7 @@ make_external_names_and_scores <- function() {
         command = {
           readRDS(external_scores_path) %>%
             group_by(forecaster) %>%
-            tar_group()
+            targets::tar_group()
         },
         iteration = "group",
         garbage_collection = TRUE

README.md

@@ -16,19 +16,24 @@ DEBUG_MODE=true
 USE_SHINY=false
 TAR_PROJECT=covid_hosp_explore
 EXTERNAL_SCORES_PATH=
-USE_AWS_S3=false
 AWS_S3_PREFIX=2023/exploration
 
+# Pull from the bucket
+make sync
+
+# Run only the dashboard, to display results run on other machines
+make dashboard
+
 # Run the pipeline wrapper run.R.
 make run
+
 ```
 
 -   `EPIDATR_USE_CACHE` controls whether `epidatr` functions use the cache.
 -   `DEBUG_MODE` controls whether `targets::tar_make` is run with the `callr_function=NULL`, which allows for debugging.
 -   `USE_SHINY` controls whether we start a Shiny server after producing the targets.
 -   `TAR_PROJECT` controls which `targets` project is run by `run.R`.
 -   `EXTERNAL_SCORES_PATH` controls where external scores are loaded from. If not set, external scores are not used.
--   `USE_AWS_S3` controls whether we use AWS S3 to store the cache.
 -   `AWS_S3_PREFIX` controls the prefix to use in the AWS S3 bucket (a prefix is a pseudo-directory in a bucket).
 
 ## Development

_targets.yaml

@@ -1,16 +1,17 @@
 covid_hosp_explore:
-    script: covid_hosp_explore.R
-    store: covid_hosp_explore
-    use_crew: yes
+  script: covid_hosp_explore.R
+  store: covid_hosp_explore
+  use_crew: yes
+  seconds_timeout: 86400
 flu_hosp_explore:
-    script: flu_hosp_explore.R
-    store: flu_hosp_explore
-    use_crew: yes
+  script: flu_hosp_explore.R
+  store: flu_hosp_explore
+  use_crew: yes
 flu_hosp_prod:
-    script: flu_hosp_prod.R
-    store: flu_hosp_prod
-    use_crew: yes
+  script: flu_hosp_prod.R
+  store: flu_hosp_prod
+  use_crew: yes
 covid_hosp_prod:
-    script: covid_hosp_prod.R
-    store: covid_hosp_prod
-    use_crew: yes
+  script: covid_hosp_prod.R
+  store: covid_hosp_prod
+  use_crew: yes

dashboard.R

@@ -0,0 +1,53 @@
+tar_project <- Sys.getenv("TAR_PROJECT", "covid_hosp_explore")
+external_scores_path <- Sys.getenv("EXTERNAL_SCORES_PATH", "")
+debug_mode <- as.logical(Sys.getenv("DEBUG_MODE", TRUE))
+use_shiny <- as.logical(Sys.getenv("USE_SHINY", FALSE))
+use_aws_s3_only <- as.logical(Sys.getenv("USE_AWS_S3_ONLY", FALSE))
+aws_s3_prefix <- Sys.getenv("AWS_S3_PREFIX", "exploration")
+aws_s3_prefix <- paste0(aws_s3_prefix, "/", tar_project)
+suppressPackageStartupMessages({
+  library(targets)
+  library(shiny)
+})
+# Prevent functions defined in /R dir from being loaded unnecessarily
+options(shiny.autoload.r = FALSE)
+
+forecaster_options <- unique(tar_read(forecaster_params_grid)[["parent_id"]])
+# Map forecaster names to score files
+forecaster_options <- setNames(
+  # File names
+  paste0("score_", gsub(" ", ".", forecaster_options)),
+  # Display names
+  forecaster_options
+)
+
+# Add ensembles
+ensemble_options <- unique(tar_read(ensemble_forecasters)[["parent_id"]])
+ensemble_options <- setNames(
+  # File names
+  paste0("ensemble_score_", ensemble_options),
+  # Display names
+  paste0("ensemble.", ensemble_options)
+)
+
+external_options <- unique(tar_read(external_names))
+EXTERNAL_PREFIX <- "[external] "
+if (!is.null(external_options) && length(external_options) > 0) {
+  external_options <- setNames(
+    # File names
+    # Get names of all branches of `external_scores` target by index. The way these
+    # were specified, `external_names` provides the order of the branches.
+    tar_branch_names(external_scores, seq_along(external_options)),
+    # Display names
+    paste0(
+      EXTERNAL_PREFIX,
+      gsub(" forecaster", "", gsub("_", " ", external_options, fixed = TRUE), fixed = TRUE)
+    )
+  )
+} else {
+  external_options <- character(0)
+}
+
+forecaster_options <- c(ensemble_options, forecaster_options, external_options)
+
+runApp(here::here("app.R"), port = 3838)

extras/targets-common.R

@@ -46,27 +46,10 @@ tar_option_set(
   # Set default crew controller.
   # https://books.ropensci.org/targets/crew.html#heterogeneous-workers
   resources = tar_resources(
-    crew = tar_resources_crew(controller = "main_controller")
+    crew = tar_resources_crew(controller = "main_controller", seconds_timeout = 24 * 60 * 60)
   )
 )
 
-use_aws_s3 <- as.logical(Sys.getenv("USE_AWS_S3", FALSE))
-tar_project <- Sys.getenv("TAR_PROJECT", "covid_hosp_explore")
-aws_s3_prefix <- Sys.getenv("AWS_S3_PREFIX", "exploration")
-aws_s3_prefix <- paste0(aws_s3_prefix, "/", tar_project)
-if (use_aws_s3) {
-  tar_option_set(
-    repository = "aws",
-    resources = tar_resources(
-      aws = tar_resources_aws(
-        bucket = "forecasting-team-data",
-        prefix = aws_s3_prefix,
-        region = "us-east-1"
-      )
-    )
-  )
-}
-
 linreg <- parsnip::linear_reg()
 quantreg <- epipredict::quantile_reg()
 ONE_AHEAD_FORECAST_NAME <- "forecast_by_ahead"