Merge pull request #1761 from cmu-delphi/ndefries/backfill/rename-fips

[Backfill corrections] Make sure pipeline has `geo_value` field to use

Author: krivard

backfill_corrections/delphiBackfillCorrection/NAMESPACE

@@ -34,6 +34,7 @@ importFrom(dplyr,group_split)
 importFrom(dplyr,if_else)
 importFrom(dplyr,mutate)
 importFrom(dplyr,pull)
+importFrom(dplyr,rename)
 importFrom(dplyr,select)
 importFrom(dplyr,summarize)
 importFrom(dplyr,ungroup)

backfill_corrections/delphiBackfillCorrection/R/io.R

@@ -1,15 +1,31 @@
 #' Read a parquet file into a dataframe
 #' 
-#' @template input_dir-template
+#' @template input_file-template
 #'
 #' @importFrom arrow read_parquet
 #'
 #' @export
-read_data <- function(input_dir) {
-  df <- read_parquet(input_dir, as_data_frame = TRUE)
+read_data <- function(input_file) {
+  df <- read_parquet(input_file, as_data_frame = TRUE)
   return (df)
 }
 
+#' Make sure data contains a `geo_value` field
+#'
+#' @template df-template
+#'
+#' @importFrom dplyr rename %>%
+#' @importFrom rlang .data
+fips_to_geovalue <- function(df) {
+  if ( !("geo_value" %in% colnames(df)) ) {
+    if ( !("fips" %in% colnames(df)) ) {
+      stop("Either `fips` or `geo_value` field must be available")
+    }
+    df <- rename(df, geo_value = .data$fips)
+  }
+  return(df)
+}
+
 #' Export the result to customized directory
 #'
 #' @param test_data test data containing prediction results

backfill_corrections/delphiBackfillCorrection/R/main.R

@@ -287,7 +287,7 @@ main <- function(params) {
     msg_ts("Reading in and combining associated files")
     input_data <- lapply(
       files_list,
-      function(file) {read_data(file)}
+      function(file) {read_data(file) %>% fips_to_geovalue()}
     ) %>%
       bind_rows()
 

backfill_corrections/delphiBackfillCorrection/R/utils.R

@@ -148,12 +148,14 @@ validity_checks <- function(df, value_type, num_col, denom_col, signal_suffixes,
     stop("No 'time_value' column detected for the reference date!")
   }
 
-  # issue_date or lag should exist in the dataset
-  if ( !lag_col %in% colnames(df) ) {
-    if ( issued_col %in% colnames(df) ) {
-      df$lag = as.integer(df$issue_date - df$time_value)
-    }
-    else {stop("No issue_date or lag exists!")}
+  # issue_date and lag should exist in the dataset
+  if ( !(lag_col %in% colnames(df)) || !(issued_col %in% colnames(df)) ) {
+    stop("`issue_date` and `lag` fields must exist in the input data")
+  }
+
+  if ( any(is.na(df[[lag_col]])) || any(is.na(df[[issued_col]])) ||
+    any(is.na(df$time_value)) ) {
+    stop("`issue_date`, `lag`, or `time_value` contain missing values")
   }
 
   return(list(df = df, value_cols = value_cols))

backfill_corrections/delphiBackfillCorrection/man-roxygen/input_file-template.R

@@ -0,0 +1 @@
+#' @param input_file path to input data file in parquet format