Merge pull request #71 from cmu-delphi/ds/minor-fix

A few minor fixes

Author: dshemetov

Makefile

@@ -3,4 +3,7 @@ install:
 	Rscript -e "renv::restore()"
 
 run:
-	Rscript run.R
+	Rscript run.R
+
+run-nohup:
+	nohup Rscript run.R &

R/targets_utils.R

@@ -61,7 +61,8 @@ lists_of_real_values <- function(param_grid) {
 #' - `hhs_signal`
 #' - `chng_signal`
 #' - `fetch_args`
-#'
+#' - `eval_time`
+#' - `traing_time`
 #' @export
 make_data_targets <- function() {
   list(
@@ -74,7 +75,7 @@ make_data_targets <- function() {
           geo_type = "state",
           time_type = "day",
           geo_values = "*",
-          time_values = epirange(from = "2020-01-01", to = "2024-01-01"),
+          time_values = eval_time,
           fetch_args = fetch_args
         )
       }
@@ -88,7 +89,7 @@ make_data_targets <- function() {
           geo_type = "state",
           time_type = "day",
           geo_values = "*",
-          time_values = epirange(from = "2020-01-01", to = "2024-01-01"),
+          time_values = eval_time,
           fetch_args = fetch_args
         )
       }
@@ -124,7 +125,7 @@ make_data_targets <- function() {
           geo_type = "state",
           time_type = "day",
           geo_values = "*",
-          time_values = epirange(from = "20220101", to = "20220401"),
+          time_values = training_time,
           issues = "*",
           fetch_args = fetch_args
         )
@@ -139,7 +140,7 @@ make_data_targets <- function() {
           geo_type = "state",
           time_type = "day",
           geo_values = "*",
-          time_values = epirange(from = "20220101", to = "20220401"),
+          time_values = training_time,
           issues = "*",
           fetch_args = fetch_args
         )

README.md

@@ -17,7 +17,7 @@ USE_SHINY=false
 TAR_PROJECT=covid_hosp_explore
 EXTERNAL_SCORES_PATH=
 USE_AWS_S3=false
-AWS_S3_PREFIX=
+AWS_S3_PREFIX=2023/exploration
 
 # Run the pipeline wrapper run.R.
 make run

_targets.yaml

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

covid_hosp_explore.R

@@ -126,7 +126,9 @@ ensembles_params_grid_target <- list(
 # during the actual targets run, since the commands are frozen as expressions).
 hhs_signal <- "confirmed_admissions_covid_1d"
 chng_signal <- "smoothed_adj_outpatient_covid"
-fetch_args <- epidatr::fetch_args_list(return_empty = TRUE, timeout_seconds = 200)
+eval_time <- epidatr::epirange(from = "2020-01-01", to = "2024-01-01")
+training_time <- epidatr::epirange(from = "2021-01-01", to = "2023-06-01")
+fetch_args <- epidatr::fetch_args_list(return_empty = TRUE, timeout_seconds = 400)
 data_targets <- make_data_targets()
 
 

extras/targets-common.R

@@ -20,7 +20,7 @@ suppressPackageStartupMessages({
 # https://books.ropensci.org/targets/crew.html#heterogeneous-workers
 main_controller <- crew_controller_local(
   name = "main_controller",
-  workers = parallel::detectCores() - 5
+  workers = parallel::detectCores() - 1L
 )
 serial_controller <- crew_controller_local(
   name = "serial_controller",
@@ -50,6 +50,23 @@ tar_option_set(
   )
 )
 
+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"

flu_hosp_explore.R

@@ -40,13 +40,14 @@ forecaster_params_grid_target <- list(
     name = forecaster_params_grid,
     command = {
       param_grid
-    }
+    },
+    priority = 0.99
   )
 )
 
 # moving on to the ensemble
 AHEADS <- 1:4
-ensemble_grid <- append(
+ensemble_grid <- add_row(
   make_shared_ensembles(),
   make_unique_ensemble_grid()
 ) %>% id_ahead_ensemble_grid(AHEADS)
@@ -71,14 +72,17 @@ ensembles_params_grid_target <- list(
     name = ensemble_forecasters,
     command = {
       ensemble_grid
-    }
+    },
+    priority = 0.99
   )
 )
 
 # These globals are needed by the function below (and they need to persist
 # during the actual targets run, since the commands are frozen as expressions).
 hhs_signal <- "confirmed_admissions_influenza_1d_prop_7dav"
 chng_signal <- "smoothed_adj_outpatient_flu"
+eval_time <- epidatr::epirange(from = "2020-01-01", to = "2024-01-01")
+training_time <- epidatr::epirange(from = "2021-01-01", to = "2023-06-01")
 fetch_args <- epidatr::fetch_args_list(return_empty = TRUE, timeout_seconds = 300)
 data_targets <- make_data_targets()
 
@@ -89,8 +93,40 @@ date_step <- 7L
 forecasts_and_scores_by_ahead <- make_forecasts_and_scores_by_ahead()
 forecasts_and_scores <- make_forecasts_and_scores()
 # ensembles
-ensemble_and_scores_by_ahead <- make_ensemble_targets_by_ahead()
-ensemble_and_scores <- make_ensemble_targets_and_scores()
+
+# do a tar_map or tar_target to asdf
+ensembles_and_scores_by_ahead <- tar_map(
+  values = target_ensemble_grid,
+  names = id,
+  tar_target(
+    name = ensemble_by_ahead,
+    command = {
+      ensemble(joined_archive_data_2022,
+        forecaster_ids,
+        "hhs",
+        extra_sources = "chng",
+        ensemble_params,
+        ensemble_params_names
+      )
+    },
+    priority = .9999
+  ),
+  tar_target(
+    name = score_by_ahead,
+    command = {
+      run_evaluation_measure(
+        data = ensemble_by_ahead,
+        evaluation_data = hhs_evaluation_data,
+        measure = list(
+          wis = weighted_interval_score,
+          ae = absolute_error,
+          cov_80 = interval_coverage(0.8)
+        )
+      )
+    }
+  )
+)
+ensembles_and_scores <- make_ensemble_targets_and_scores()
 # other sources
 external_names_and_scores <- make_external_names_and_scores()
 

man/make_data_targets.Rd

@@ -32,6 +32,7 @@ debug_mode <- as.logical(Sys.getenv("DEBUG_MODE", TRUE))
 use_shiny <- as.logical(Sys.getenv("USE_SHINY", FALSE))
 use_aws_s3 <- as.logical(Sys.getenv("USE_AWS_S3", FALSE))
 aws_s3_prefix <- Sys.getenv("AWS_S3_PREFIX", "exploration")
+aws_s3_prefix <- paste0(aws_s3_prefix, "/", tar_project)
 cli::cli_inform(
   c(
     "i" = "Reading environment variables...",
@@ -53,17 +54,6 @@ suppressPackageStartupMessages({
 store_dir <- tar_path_store()
 if (!dir.exists(store_dir)) dir.create(store_dir)
 
-if (use_aws_s3) {
-  tar_option_set(
-    repository = "aws",
-    resources = tar_resources(
-      aws = tar_resources_aws(
-        bucket = "forecasting-team-data",
-        prefix = aws_s3_prefix
-      )
-    )
-  )
-}
 
 tar_manifest()
 if (debug_mode) {
@@ -78,7 +68,7 @@ if (use_shiny) {
   # Prevent functions defined in /R dir from being loaded unnecessarily
   options(shiny.autoload.r = FALSE)
 
-  forecaster_options <- unique(tar_read(forecasters)[["parent_id"]])
+  forecaster_options <- unique(tar_read(forecaster_params_grid)[["parent_id"]])
   # Map forecaster names to score files
   forecaster_options <- setNames(
     # File names