Merge pull request #1572 from cmu-delphi/ndefries/add-weekly-weights-microdata

[CTIS] Add weekly weights to microdata

Author: krivard

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

@@ -57,6 +57,7 @@
   "start_date": "2021-08-16",
   "static_dir": "./static",
   "weights_in_dir": "./fb-incoming",
+  "weekly_weights_in_dir": "./fb-incoming-weekly",
   "weights_out_dir": "./fb-outgoing",
   "experimental_weights_out_dir": "./exp-fb-outgoing"
 }

facebook/Makefile

@@ -5,12 +5,16 @@ TODAY:=$(shell date +"%Y-%m-%d")
 YESTERDAY:=$(shell date --date "$(TODAY) -1 day" +"%Y-%m-%d")
 ONEWEEK:=$(shell date --date "$(TODAY) -7 day" +"%Y-%m-%d")
 THREEWEEK:=$(shell date --date "$(TODAY) -21 day" +"%Y-%m-%d")
+LAST_SATURDAY:=$(shell date -d "last Saturday" +"%Y-%m-%d")
+LAST_SUNDAY:=$(shell date -d "$(LAST_SATURDAY) -6 day" +"%Y-%m-%d")
+TUESDAY:=$(shell date -d "$(LAST_SATURDAY) +3 day" +"%Y-%m-%d")
 
 MESSAGES:="messages/$(TODAY).messages"
 
 PYTHON:=env/bin/python
 QUALTRICS=$(shell $(PYTHON) -m delphi_utils get input_dir)
 WEIGHTS=$(shell $(PYTHON) -m delphi_utils get weights_in_dir)
+WEEKLY_WEIGHTS=$(shell $(PYTHON) -m delphi_utils get weekly_weights_in_dir)
 CIDS=$(shell $(PYTHON) -m delphi_utils get weights_out_dir)
 CIDS_EXP=$(shell $(PYTHON) -m delphi_utils get experimental_weights_out_dir)
 INDIVIDUAL=$(shell $(PYTHON) -m delphi_utils get individual_dir)
@@ -24,6 +28,7 @@ DELPHI_SURVEY_EMAIL_USER=$(shell $(PYTHON) -m delphi_utils get delphi_survey_ema
 SFTP_OPTIONS=$(shell $(PYTHON) -m delphi_utils get sftp_options)
 
 MAX_WEIGHTED=ls -1 $(WEIGHTS) | grep dap | tail -1 | sed 's/_.*//;s/-//g;'
+MAX_WEEKLY_WEIGHTED=ls -1 $(WEEKLY_WEIGHTS) | grep map | tail -1 | sed 's/_.*//;s/-//g;'
 
 ANTIJOIN:="antijoin.cids.sorted.txt"
 ANTIJOIN_EXP:="antijoin.experimental.cids.sorted.txt"
@@ -42,24 +47,35 @@ else
 	SFTP_POST:=sshpass -p $(DELPHI_SURVEY_SFTP_PASSWORD) sftp $(SFTP_OPTIONS) -b <(echo -e "$${BATCH}") -P 2222 $(DELPHI_SURVEY_SFTP_USER)
 endif
 
+ifneq ("$(wildcard params.json)","")
+ifeq ($(WEIGHTS),$(WEEKLY_WEIGHTS))
+$(error "'weights_in_dir' and 'weekly_weights_in_dir' must be different.")
+endif
+endif
+
 default:
 	@echo No default implemented yet
 
 scratch:
 	mkdir scratch
 	rm -rf scratch/*
 
-tidy: receiving
-	rm -rf tidy/$(RECEIVING)
-	rm -rf tidy/$(INDIVIDUAL)
-	rm -f tidy/params.json
-	mkdir -p tidy tidy/$(RECEIVING) tidy/$(INDIVIDUAL)
-	cp params.json tidy/
-	mv $(RECEIVING)/*.csv tidy/$(RECEIVING)
-	mv $(INDIVIDUAL)/*.csv* tidy/$(INDIVIDUAL)
-	mv $(INDIVIDUAL_RACEETH)/*.csv* tidy/$(INDIVIDUAL_RACEETH)
-	tar -czf scratch/tidy-`date +"%Y-%m-%d-%H%M%S"`.tgz --exclude='tidy-*.tgz' tidy
-	mv scratch/*.tgz tidy/
+$(INDIVIDUAL) $(INDIVIDUAL_RACEETH):
+	mkdir $@
+
+tidy_%: receiving
+	rm -rf $@/$(RECEIVING)
+	rm -rf $@/$(INDIVIDUAL)
+	rm -rf $@/$(INDIVIDUAL_RACEETH)
+	rm -f $@/params.json
+	mkdir -p $@ $@/$(RECEIVING) $@/$(INDIVIDUAL) $@/$(INDIVIDUAL_RACEETH)
+	cp params.json $@/
+	# Check for _any_ matching files using https://stackoverflow.com/a/6364244/14401472
+	if compgen -G "$(RECEIVING)/*.csv" > /dev/null; then mv $(RECEIVING)/*.csv $@/$(RECEIVING); fi
+	mv $(INDIVIDUAL)/*.csv* $@/$(INDIVIDUAL)
+	mv $(INDIVIDUAL_RACEETH)/*.csv* $@/$(INDIVIDUAL_RACEETH)
+	tar -czf scratch/$@-`date +"%Y-%m-%d-%H%M%S"`.tgz --exclude='tidy*-*.tgz' --exclude='*.done' $@
+	mv scratch/*.tgz $@/
 
 clean:
 	rm -f $(RECEIVING)/*.csv $(INDIVIDUAL)/*.csv $(INDIVIDUAL_RACEETH)/*.csv $(CIDS)/*.csv $(CIDS_EXP)/*.csv
@@ -114,6 +130,17 @@ params.json: $(TODAY)
 		output cids,individual,covidalert,archive,community \
 		start_date $(YESTERDAY)
 
+params.weekly-weights.json: $(TODAY)
+	PAT=`grep fb-survey params.json | awk 'BEGIN{FS="\""}{print $$2}' | sed 's/ /_/g;s/^/-e /'`; \
+	$(PYTHON) -m delphi_utils set \
+		debug false \
+		produce_individual_raceeth true \
+		end_date $(LAST_SATURDAY) \
+		input <(find $(QUALTRICS) -maxdepth 1 -newer $< -type f -name "*.csv" | sort | grep $${PAT}  | tr '\n' ',' | sed 's_$(QUALTRICS)/__g;s/,$$//' ) \
+		parallel true \
+		output individual \
+		start_date $(LAST_SUNDAY)
+
 $(WEIGHTS): $(TODAY)
 	[ -f $(WEIGHTS) ] || mkdir -p $(WEIGHTS)
 	cd "$(WEIGHTS)"; \
@@ -133,6 +160,40 @@ $(WEIGHTS): $(TODAY)
 	  echo "WARNING: $${MSG}" >> $(MESSAGES); \
 	fi
 
+$(WEEKLY_WEIGHTS): $(TODAY)
+# This runs every day as a dependency of `pipeline`. A pipeline run is triggered when new weekly weights files are available.
+	[ -f $(WEEKLY_WEIGHTS) ] || mkdir -p $(WEEKLY_WEIGHTS)
+	cd "$(WEEKLY_WEIGHTS)"; \
+	BATCH="cd fb-interchange/cmu_respondent_ww_weights\nls -1"; \
+	NEW=`LC_ALL=C comm -23 <(sshpass -p $(DELPHI_SURVEY_SFTP_PASSWORD) sftp $(SFTP_OPTIONS) -b <(echo -e "$${BATCH}") -P 2222 $(DELPHI_SURVEY_SFTP_USER) | grep "^202" | LC_ALL=C sort) <(ls -1 | LC_ALL=C sort)`; \
+	echo "New weekly weights files:"; \
+	echo $${NEW}; \
+	for f in $${NEW}; do \
+	  BATCH="$${BATCH}\nget $$f"; \
+	done; \
+	sshpass -p $(DELPHI_SURVEY_SFTP_PASSWORD) sftp $(SFTP_OPTIONS) -b <(echo -e "$${BATCH}") -P 2222 $(DELPHI_SURVEY_SFTP_USER) || exit 90; \
+	cd -; \
+	touch -d $(YESTERDAY) $(WEEKLY_WEIGHTS); \
+	EXPECTED_WEEKLY_WEIGHTED=`date --date='$(LAST_SUNDAY)' +'%Y%m%d'`; \
+	MIN_NEW_WEEKLY_WEIGHTED=`grep map <<< $${NEW} | head -1 | sed 's/_.*//;s/-//g;'`; \
+	if [[ `wc -w <<< $${NEW}` -gt 0 ]] && [[ $$MIN_NEW_WEEKLY_WEIGHTED -ne $$EXPECTED_WEEKLY_WEIGHTED ]]; then \
+	  MSG="Expected new weekly weights files to start on: $$EXPECTED_WEEKLY_WEIGHTED; Actual new files starts on: $$MIN_NEW_WEEKLY_WEIGHTED"; \
+	  echo "WARNING: $${MSG}" >> $(MESSAGES); \
+	fi; \
+	MAX_WEEKLY_WEIGHTED=`$(MAX_WEEKLY_WEIGHTED)`; \
+	if [[ `date --date='$(TODAY)' +'%Y%m%d'` -gt `date --date='$(TUESDAY)' +'%Y%m%d'` ]] && [[ $$MAX_WEEKLY_WEIGHTED -lt $$EXPECTED_WEEKLY_WEIGHTED ]]; then \
+	  MSG="Weekly weights are old; Expected most recent weekly weights file to start on: $$EXPECTED_WEEKLY_WEIGHTED; Actual most recent file starts on: $$MAX_WEEKLY_WEIGHTED"; \
+	  echo "WARNING: $${MSG}" >> $(MESSAGES); \
+	fi; \
+	if [[ ! -f tidy_weekly/$(LAST_SUNDAY)-weekly-weights.done ]] && [[ $$MAX_WEEKLY_WEIGHTED -eq $$EXPECTED_WEEKLY_WEIGHTED ]]; then \
+	  if [ -f params.json ]; then cp params.json params.daily.json; fi; \
+	  $(MAKE) weekly-weights-pipeline; \
+	  if [ -f params.daily.json ]; then \
+		cp params.daily.json params.json; \
+		rm -f params.daily.json; \
+	  fi; \
+	fi
+
 dev: delphiFacebook_1.0.tar.gz
 	R CMD INSTALL delphiFacebook_1.0.tar.gz
 
@@ -146,13 +207,18 @@ run-R: $(CIDS) $(CIDS_EXP)
 	grep "scheduled core" tmp ; \
 	[ "$$?" -eq 1 ]
 
-pipeline: scratch init-qualtrics params.json $(WEIGHTS) run-R post-cids post-experimental-cids post-individual post-individual-raceeth post-done tidy
+pipeline: scratch init-qualtrics params.json $(WEIGHTS) run-R post-cids post-experimental-cids post-individual post-individual-raceeth post-done tidy_daily $(WEEKLY_WEIGHTS)
 	grep $(TODAY) params.json
 	[ -f $(YESTERDAY) ] && rm $(YESTERDAY) || true
 	touch $@
 	echo "SUCCESS: $(DRY_MESSAGE)pipeline complete" >> $(MESSAGES)
 	chmod o+w $(MESSAGES)
 
+weekly-weights-pipeline: scratch init-qualtrics params.weekly-weights.json run-R post-individual post-individual-raceeth tidy_weekly
+	touch $@
+	echo "SUCCESS: $(DRY_MESSAGE)completed weekly weights pipeline" >> $(MESSAGES)
+	touch tidy_weekly/$(LAST_SUNDAY)-weekly-weights.done
+
 coverage:
 	Rscript -e 'covr::package_coverage("delphiFacebook")'
 

facebook/delphiFacebook/NAMESPACE

@@ -1,5 +1,6 @@
 # Generated by roxygen2: do not edit by hand
 
+export(add_weights)
 export(apply_privacy_censoring)
 export(assert)
 export(ceiling_epiweek)
@@ -21,7 +22,6 @@ export(get_range_prev_full_period)
 export(get_range_prev_full_week)
 export(get_sparse_filenames)
 export(jeffreys_se)
-export(join_weights)
 export(load_archive)
 export(load_response_one)
 export(load_responses_all)

facebook/delphiFacebook/R/contingency_run.R

@@ -144,7 +144,7 @@ run_contingency_tables_one_period <- function(params, aggregations)
       return()
     }
 
-    data_agg <- join_weights(data_agg, params, weights = "full")$df
+    data_agg <- add_weights(data_agg, params, weights = "full")$df
     msg_df("response data to aggregate", data_agg)
 
     produce_aggregates(data_agg, aggregations, cw_list, params)

facebook/delphiFacebook/R/responses.R

@@ -779,10 +779,10 @@ filter_complete_responses <- function(data_full, params)
   data_full <- select(data_full, -.data$zip5)
 
   # 9 includes StartDatetime, EndDatetime, Date, token, wave, geo_id,
-  # UserLanguage + two questions (ignore raceethnicity, module, and
-  # w12_assignment fields which may or may not exist, depending on params and
+  # UserLanguage + two questions (ignore raceethnicity, module,
+  # w12_assignment, and weekly weights fields which may or may not exist, depending on params and
   # survey version)
-  ignore_cols <- c("raceethnicity", "w12_assignment", "module")
+  ignore_cols <- c("raceethnicity", "w12_assignment", "module", "weight_wf", "weight_wp")
   valid_row_filter <- rowSums( !is.na(data_full[, !(names(data_full) %in% ignore_cols)]) ) >= 9
   data_full <- data_full[valid_row_filter, ]
 

facebook/delphiFacebook/R/run.R

@@ -27,7 +27,7 @@ run_facebook <- function(params)
   # create data that will be aggregated for covidcast
   data_agg <- create_data_for_aggregation(input_data)
   data_agg <- filter_data_for_aggregation(data_agg, params, lead_days = 12)
-  weight_result <- join_weights(data_agg, params, weights = "step1")
+  weight_result <- add_weights(data_agg, params, weights = "step1")
   data_agg <- weight_result$df
   latest_weight_date_step1 <- weight_result$weight_date
   msg_df("response data to aggregate", data_agg)
@@ -36,10 +36,10 @@ run_facebook <- function(params)
     is.na(latest_weight_date_step1), as.Date(params$end_date), latest_weight_date_step1
   )
 
-  # create "complete" data that will be shared with research partners
+  # create "complete" data (microdata) that will be shared with research partners
   data_full <- create_complete_responses(input_data, cw_list$county, params)
   data_full <- filter_complete_responses(data_full, params)
-  data_full <- join_weights(data_full, params, weights = "full")$df
+  data_full <- add_weights(data_full, params, weights = "full", add_weekly_weights = TRUE)$df
   msg_df("full data to share with research partners", data_full)
 
   # create module-complete data used to create CID lists separately for each module

facebook/delphiFacebook/R/weights.R

@@ -83,17 +83,15 @@ generate_cid_list_filename <- function(type_name, params, module_type) {
 #' exclusively for producing individual response files.
 #'
 #' @param data    a data frame containing a column called "token"
-#' @param params  a named list containing a value "weights_in_dir" indicating where the
-#'                weights files are stored
+#' @param params  a named list containing value "weights_in_dir", indicating where the
+#'                weights files are stored, and "weekly_weights_in_dir", indicating
+#'                where the weekly weights files are stored, if `add_weekly_weights` is TRUE
 #' @param weights Which weights to use -- step1 or full?
+#' @param add_weekly_weights boolean indicating whether to add weekly partial and full
+#'                weights in addition to daily weights
 #'
-#' @importFrom dplyr bind_rows left_join
-#' @importFrom data.table fread
-#' @importFrom stringi stri_extract_first
-#' @importFrom utils tail
-#' 
 #' @export
-join_weights <- function(data, params, weights = c("step1", "full"))
+add_weights <- function(data, params, weights = c("step1", "full"), add_weekly_weights = FALSE)
 {
   weights <- match.arg(weights)
 
@@ -103,14 +101,47 @@ join_weights <- function(data, params, weights = c("step1", "full"))
     pattern <- "finish_full_survey_weights.csv$"
   }
 
-  weights_files <- dir(params$weights_in_dir, pattern = pattern, full.names = TRUE)
+  weight_result <- load_and_join_weights(data, params$weights_in_dir, pattern)
+  data <- weight_result$df
+  latest_weight_date <- weight_result$weight_date
+
+  if (add_weekly_weights) {
+    # Since each weight column is joined on as `weight`, we need to rename
+    # each new weight column before performing the next join to avoid
+    # overwriting any weights
+    data <- rename(data, daily_weight = weight)
+    data <- load_and_join_weights(
+        data, params$weekly_weights_in_dir, pattern = "partial_weekly_weights.csv$"
+      )$df %>%
+      rename(weight_wp = weight)
+    data <- load_and_join_weights(
+        data, params$weekly_weights_in_dir, pattern = "full_weekly_weights.csv$"
+      )$df %>%
+      rename(weight_wf = weight, weight = daily_weight)
+  }
+
+  return( list(df = data, weight_date = latest_weight_date) )
+}
+
+#' Add a single type of weights to a dataset of responses as field `weight`
+#'
+#' @param data    a data frame containing a column called "token"
+#' @param weights_dir directory to look for the weights in
+#' @param pattern regular expression matching desired weights files
+#'
+#' @importFrom dplyr bind_rows left_join
+#' @importFrom data.table fread
+#' @importFrom stringi stri_extract_first
+#' @importFrom utils tail
+load_and_join_weights <- function(data, weights_dir, pattern) {
+  weights_files <- dir(weights_dir, pattern = pattern, full.names = TRUE)
   weights_files <- sort(weights_files)
 
   latest_weight <- tail(weights_files, n = 1)
   latest_weight_date <- as.Date(
     stri_extract_first(basename(latest_weight), regex = "^[0-9]{4}-[0-9]{2}-[0-9]{2}")
   )
-  
+
   col_types <- c("character", "double")
   col_names <- c("cid", "weight")
   agg_weights <- bind_rows(lapply(
@@ -122,6 +153,6 @@ join_weights <- function(data, params, weights = c("step1", "full"))
   )
   agg_weights <- agg_weights[!duplicated(cid),]
   data <- left_join(data, agg_weights, by = c("token" = "cid"))
-
+  
   return( list(df = data, weight_date = latest_weight_date) )
 }

facebook/delphiFacebook/integration-tests/testthat/params-full.json

@@ -9,6 +9,7 @@
   "archive_dir": "./archive",
   "individual_dir": "./individual_full",
   "weights_in_dir": "./weights_full",
+  "weekly_weights_in_dir": "./weights_full_weekly",
   "weights_out_dir": "./weights_out",
   "experimental_weights_out_dir": "./exp_weights_out",
   "input_dir": "./input",

facebook/delphiFacebook/integration-tests/testthat/params-test.json

@@ -9,6 +9,7 @@
   "archive_dir": "./archive",
   "individual_dir": "./individual",
   "weights_in_dir": "./weights_in",
+  "weekly_weights_in_dir": "./weights_in_weekly",
   "weights_out_dir": "./weights_out",
   "experimental_weights_out_dir": "./exp_weights_out",
   "input_dir": "./input",

facebook/delphiFacebook/integration-tests/testthat/test-contingency-run.R

@@ -161,7 +161,7 @@ set.seed(0)
 rand_weights <- runif(2000)
 rand_weights <- rand_weights / sum(rand_weights)
 
-mock_join_weights <- function(data, params, weights = c("step1", "full")) {
+mock_add_weights <- function(data, params, weights = c("step1", "full")) {
   data <- cbind(as.data.table(data), weight=rand_weights)
   return( list(df = data, weight_date = NA) )
 }
@@ -190,7 +190,7 @@ test_that("simple weighted dataset produces correct percents", {
   params <- get_params(tdir)
   create_dir_not_exist(params$export_dir)
 
-  local_mock("delphiFacebook::join_weights" = mock_join_weights)
+  local_mock("delphiFacebook::add_weights" = mock_add_weights)
   local_mock("delphiFacebook::mix_weights" = mock_mix_weights)
   local_mock("delphiFacebook::load_archive" = mock_load_archive)
   local_mock("delphiFacebook::add_geo_vars" = mock_geo_vars)

facebook/delphiFacebook/integration-tests/testthat/test-gold-individual.R

@@ -20,7 +20,8 @@ test_that("individual output files exist", {
 test_that("contents are the same", {
   for (f in expected_files) {
     test_df <- read.csv(test_path("individual_full", f))
-    gold_df <- read.csv(test_path("gold_individual", f))
+    gold_df <- read.csv(test_path("gold_individual", f)) %>%
+      mutate(weight_wp = weight, weight_wf = weight)
 
     expect_equal(nrow(gold_df), nrow(test_df))
 

facebook/delphiFacebook/integration-tests/testthat/test-integration.R

@@ -152,7 +152,7 @@ test_that("testing weighted community values files", {
   params <- relativize_params(read_params(test_path("params-test.json")))
 
   input_data <- load_responses_all(params)
-  input_data <- join_weights(input_data, params)$df
+  input_data <- add_weights(input_data, params)$df
 
   # there are 2 / 4 households in PA on 2020-05-11 for community
   these <- input_data[input_data$date == "2020-05-11" & input_data$zip5 == "15106",]
@@ -194,7 +194,7 @@ test_that("testing weighted smoothed community values files", {
   params <- relativize_params(read_params(test_path("params-test.json")))
 
   input_data <- load_responses_all(params)
-  input_data <- join_weights(input_data, params)$df
+  input_data <- add_weights(input_data, params)$df
 
   # there are 2 / 4 households in PA on 2020-05-11 for community
   these <- input_data[
@@ -328,7 +328,7 @@ test_that("testing weighted ili/cli values files", {
   params <- relativize_params(read_params(test_path("params-test.json")))
 
   input_data <- load_responses_all(params)
-  input_data <- join_weights(input_data, params)$df
+  input_data <- add_weights(input_data, params)$df
   data_agg <- create_data_for_aggregation(input_data)
 
   ## There are 4 households in PA on 2020-05-11, one with ILI.