merge in main

merge in main

# Conflicts:
#	renv.lock

R/aws_s3_upload_single_type.R

@@ -0,0 +1,22 @@
+#' .. content for \description{} (no empty lines) ..
+#'
+#' .. content for \details{} ..
+#'
+#' @title
+#' @param path
+#' @param bucket
+#' @param key
+#' @param check
+#' @return
+#' @author Emma Mendelsohn
+#' @export
+aws_s3_upload_single_type <- function(directory_path,
+                                      bucket, 
+                                      key, 
+                                      check = TRUE) {
+
+  file.remove(file.path(directory_path, ".gitkeep"))
+  containerTemplateUtils::aws_s3_upload(path = directory_path,bucket =  bucket,key =  key, check = check)
+  file.create(file.path(directory_path, ".gitkeep"))
+
+}

R/calculate_ndvi_anomalies.R

@@ -0,0 +1,83 @@
+#' .. content for \description{} (no empty lines) ..
+#'
+#' .. content for \details{} ..
+#'
+#' @title
+#' @param ndvi_date_lookup
+#' @param ndvi_historical_means
+#' @param ndvi_anomalies_directory
+#' @param model_dates
+#' @param model_dates_selected
+#' @param lag_intervals
+#' @param overwrite
+#' @return
+#' @author Emma Mendelsohn
+#' @export
+calculate_ndvi_anomalies <- function(ndvi_date_lookup, ndvi_historical_means,
+                                     ndvi_anomalies_directory, model_dates,
+                                     model_dates_selected, lag_intervals,
+                                     overwrite = FALSE) {
+
+  # Set filename
+  date_selected <- model_dates_selected
+  save_filename <- glue::glue("ndvi_anomaly_{date_selected}.gz.parquet")
+  message(paste0("Calculating NDVI anomalies for ", date_selected))
+
+  # Check if file already exists
+  existing_files <- list.files(ndvi_anomalies_directory)
+  if(save_filename %in% existing_files & !overwrite) {
+    message("file already exists, skipping download")
+    return(file.path(ndvi_anomalies_directory, save_filename))
+  }
+
+  # Get historical means for DOY
+  doy <- model_dates |> filter(date == date_selected) |> pull(day_of_year)
+  doy_frmt <- str_pad(doy,width = 3, side = "left", pad = "0")
+  historical_means <- read_parquet(ndvi_historical_means[str_detect(ndvi_historical_means, doy_frmt)]) |> 
+    select(-day_of_year)
+
+  # Get the lagged anomalies for selected dates, mapping over the lag intervals
+  row_select <- which(model_dates$date == date_selected)
+
+  lag_intervals_start <- c(1 , 1+lag_intervals[-length(lag_intervals)])
+  lag_intervals_end <- lag_intervals
+
+  anomalies <- map2(lag_intervals_start, lag_intervals_end, function(start, end){
+
+    lag_dates <- model_dates |> slice((row_select - start):(row_select - end))
+
+    # get files and weights for the calculations
+    weights <- ndvi_date_lookup |> 
+      mutate(lag_date = map(lookup_dates, ~. %in% lag_dates$date)) |> 
+      mutate(weight = unlist(map(lag_date, sum))) |> 
+      filter(weight > 0) |> 
+      select(start_date, filename, weight)
+
+   ndvi_dataset <- open_dataset(weights$filename)
+
+   # Lag: calculate mean by pixel for the preceding x days
+   lagged_means <- ndvi_dataset |> 
+     left_join(weights |> select(-filename))  |> 
+        group_by(x, y) |>
+        summarize(lag_ndvi_mean = sum(ndvi * weight)/ sum(weight)) |>
+        ungroup()
+
+      # Join in historical means to calculate anomalies raw and scaled
+      full_join(lagged_means, historical_means, by = c("x", "y")) |>
+        mutate(!!paste0("anomaly_ndvi_", end) := lag_ndvi_mean - historical_ndvi_mean,
+               !!paste0("anomaly_ndvi_scaled_", end) := (lag_ndvi_mean - historical_ndvi_mean)/historical_ndvi_sd) |>
+        select(-starts_with("lag"), -starts_with("historical"))
+    }) |>
+      reduce(left_join, by = c("x", "y")) |>
+      mutate(date = date_selected) |>
+      relocate(date)
+
+    # Save as parquet 
+    write_parquet(anomalies, here::here(ndvi_anomalies_directory, save_filename), compression = "gzip", compression_level = 5)
+
+    return(file.path(ndvi_anomalies_directory, save_filename))
+
+
+
+  }
+

R/calculate_ndvi_historical_means.R

@@ -0,0 +1,53 @@
+#' .. content for \description{} (no empty lines) ..
+#'
+#' .. content for \details{} ..
+#'
+#' @title
+#' @param sentinel_ndvi_transformed
+#' @param sentinel_ndvi_transformed_directory
+#' @param modis_ndvi_transformed
+#' @param modis_ndvi_transformed_directory
+#' @param ndvi_date_lookup
+#' @param days_of_year
+#' @param overwrite
+#' @return
+#' @author Emma Mendelsohn
+#' @export
+calculate_ndvi_historical_means <- function(ndvi_historical_means_directory,
+                                            ndvi_date_lookup, days_of_year,
+                                            overwrite = FALSE) {
+
+  # Set filename
+  doy <- days_of_year
+  doy_frmt <- str_pad(doy,width = 3, side = "left", pad = "0")
+  save_filename <- glue::glue("historical_ndvi_mean_doy_{doy_frmt}.gz.parquet")
+  message(paste("calculating historical ndvi means and standard deviations for doy", doy_frmt))
+
+  # Check if file already exists
+  existing_files <- list.files(ndvi_historical_means_directory)
+  if(save_filename %in% existing_files & !overwrite) {
+    message("file already exists, skipping download")
+    return(file.path(ndvi_historical_means_directory, save_filename))
+  }
+
+  # Get relevant NDVI intervals
+  doy_lookup <-  ndvi_date_lookup |> 
+    filter(map_lgl(lookup_day_of_year, ~any(. == doy)))
+
+  # Create dataset of relevant files
+  ndvi_dataset <- open_dataset(doy_lookup$filename)
+
+  # Calculate historical means and standard deviations
+  historical_means <- ndvi_dataset |> 
+    mutate(day_of_year = doy) |> 
+    group_by(x, y, day_of_year) |> 
+    summarize(historical_ndvi_mean = mean(ndvi),
+              historical_ndvi_sd = sd(ndvi)) |> 
+    ungroup() 
+
+  # Save as parquet 
+  write_parquet(historical_means, here::here(ndvi_historical_means_directory, save_filename), compression = "gzip", compression_level = 5)
+
+  return(file.path(ndvi_historical_means_directory, save_filename))
+
+}

R/calculate_weather_anomalies.R

@@ -0,0 +1,79 @@
+#' .. content for \description{} (no empty lines) ..
+#'
+#' .. content for \details{} ..
+#'
+#' @title
+#' @param sentinel_ndvi_transformed
+#' @param nasa_weather_transformed
+#' @return
+#' @author Emma Mendelsohn
+#' @export
+calculate_weather_anomalies <- function(nasa_weather_transformed,
+                                        nasa_weather_transformed_directory,
+                                        weather_historical_means,
+                                        weather_anomalies_directory,
+                                        model_dates,
+                                        model_dates_selected,
+                                        lag_intervals,
+                                        overwrite = FALSE) {
+
+  # Set filename
+  date_selected <- model_dates_selected
+  save_filename <- glue::glue("weather_anomaly_{date_selected}.gz.parquet")
+  message(paste0("Calculating weather anomalies for ", date_selected))
+
+  # Check if file already exists
+  existing_files <- list.files(weather_anomalies_directory)
+  if(save_filename %in% existing_files & !overwrite) {
+    message("file already exists, skipping download")
+    return(file.path(weather_anomalies_directory, save_filename))
+  }
+
+  # Open dataset to transformed data
+  weather_transformed_dataset <- open_dataset(nasa_weather_transformed_directory)
+
+  # Get historical means for DOY
+  doy <- model_dates |> filter(date == date_selected) |> pull(day_of_year)
+  doy_frmt <- str_pad(doy,width = 3, side = "left", pad = "0")
+  historical_means <- read_parquet(weather_historical_means[str_detect(weather_historical_means, doy_frmt)]) |> 
+    select(-day_of_year)
+
+  # Get the lagged anomalies for selected dates, mapping over the lag intervals
+  row_select <- which(model_dates$date == date_selected)
+
+  lag_intervals_start <- c(1 , 1+lag_intervals[-length(lag_intervals)])
+  lag_intervals_end <- lag_intervals
+
+  anomalies <- map2(lag_intervals_start, lag_intervals_end, function(start, end){
+    lag_dates <- model_dates |> slice((row_select - start):(row_select - end))
+
+    # Lag: calculate mean by pixel for the preceding x days
+    lagged_means <- weather_transformed_dataset |> 
+      filter(date %in% !!lag_dates$date) |> 
+      group_by(x, y) |> 
+      summarize(lag_relative_humidity_mean = mean(relative_humidity),
+                lag_temperature_mean = mean(temperature),
+                lag_precipitation_mean = mean(precipitation)) |> 
+      ungroup() 
+
+    # Join in historical means to calculate anomalies raw and scaled
+    full_join(lagged_means, historical_means, by = c("x", "y")) |> 
+      mutate(!!paste0("anomaly_relative_humidity_", end) := lag_relative_humidity_mean - historical_relative_humidity_mean,
+             !!paste0("anomaly_temperature_", end) := lag_temperature_mean  -  historical_temperature_mean,
+             !!paste0("anomaly_precipitation_", end) := lag_precipitation_mean - historical_precipitation_mean,
+             !!paste0("anomaly_relative_humidity_scaled_", end) := (lag_relative_humidity_mean - historical_relative_humidity_mean)/historical_relative_humidity_sd,
+             !!paste0("anomaly_temperature_scaled_", end) := (lag_temperature_mean  -  historical_temperature_mean)/historical_temperature_sd,
+             !!paste0("anomaly_precipitation_scaled_", end) := (lag_precipitation_mean - historical_precipitation_mean)/historical_precipitation_sd) |> 
+      select(-starts_with("lag"), -starts_with("historical"))
+  }) |> 
+    reduce(left_join, by = c("x", "y")) |> 
+    mutate(date = date_selected) |> 
+    relocate(date)
+
+  # Save as parquet 
+  write_parquet(anomalies, here::here(weather_anomalies_directory, save_filename), compression = "gzip", compression_level = 5)
+
+  return(file.path(weather_anomalies_directory, save_filename))
+
+
+}

R/calculate_weather_historical_means.R

@@ -0,0 +1,51 @@
+#' .. content for \description{} (no empty lines) ..
+#'
+#' .. content for \details{} ..
+#'
+#' @title
+#' @param nasa_weather_transformed
+#' @param nasa_weather_transformed_directory
+#' @param weather_historical_means_directory
+#' @return
+#' @author Emma Mendelsohn
+#' @export
+calculate_weather_historical_means <- function(nasa_weather_transformed,
+                                               nasa_weather_transformed_directory,
+                                               weather_historical_means_directory,
+                                               days_of_year,
+                                               overwrite = FALSE) {
+
+  # Set filename
+  doy <- days_of_year
+  doy_frmt <- str_pad(doy,width = 3, side = "left", pad = "0")
+  save_filename <- glue::glue("historical_weather_mean_doy_{doy_frmt}.gz.parquet")
+  message(paste("calculating historical weather means and standard deviations for doy", doy_frmt))
+
+  # Check if file already exists
+  existing_files <- list.files(weather_historical_means_directory)
+  if(save_filename %in% existing_files & !overwrite) {
+    message("file already exists, skipping download")
+    return(file.path(weather_historical_means_directory, save_filename))
+  }
+  # Open dataset to transformed data
+  weather_transformed_dataset <- open_dataset(nasa_weather_transformed_directory)
+
+  # Filter for day of year and calculate historical means and standard deviations
+  historical_means <- weather_transformed_dataset |> 
+    filter(day_of_year == doy) |> 
+    group_by(x, y, day_of_year) |> 
+    summarize(historical_relative_humidity_mean = mean(relative_humidity),
+              historical_temperature_mean = mean(temperature),
+              historical_precipitation_mean = mean(precipitation),
+              historical_relative_humidity_sd = sd(relative_humidity),
+              historical_temperature_sd = sd(temperature),
+              historical_precipitation_sd = sd(precipitation)) |> 
+    ungroup() 
+
+  # Save as parquet 
+  write_parquet(historical_means, here::here(weather_historical_means_directory, save_filename), compression = "gzip", compression_level = 5)
+
+  return(file.path(weather_historical_means_directory, save_filename))
+
+
+}