Switch to parallelly and made functions hpc-friendly

DESCRIPTION

@@ -27,7 +27,7 @@ Description: What the package does (one paragraph).
 License: GPL (>= 3)
 Encoding: UTF-8
 Roxygen: list(markdown = TRUE)
-RoxygenNote: 7.3.1
+RoxygenNote: 7.3.2
 Suggests: 
     here,
     knitr,
@@ -42,7 +42,7 @@ Imports:
     lubridate,
     magrittr,
     ncdf4,
-    parallel,
+    parallelly,
     purrr,
     stringr,
     terra

NAMESPACE

@@ -7,8 +7,12 @@ export(htr_change_freq)
 export(htr_create_ensemble)
 export(htr_download_ESM)
 export(htr_fix_calendar)
+export(htr_integrate_levels)
 export(htr_make_folder)
 export(htr_merge_files)
 export(htr_regrid_esm)
+export(htr_seasonal_frequency)
+export(htr_shift_years)
+export(htr_show_levels)
 export(htr_slice_period)
 importFrom(magrittr,"%>%")

R/htr_change_freq.R

@@ -7,7 +7,16 @@
 #' @export
 #'
 #' @examples
-htr_change_freq <- function(freq,
+#' htr_change_freq(
+#' hpc = NA,
+#' file = NA,
+#' freq = "monthly",
+#' indir = here("data", "proc", "sliced", variable),
+#' outdir = here("data", "proc", "monthly", variable)
+#' )
+htr_change_freq <- function(hpc = NA, # if ran in the HPC, possible values are "array", "parallel"
+                            file = NA, # hpc = "array", the input will be the file
+                            freq, # possible values are "yearly" or "monthly"
                             indir,
                             outdir) {
   . <- NULL # Stop devtools::check() complaints about NSE
@@ -16,9 +25,12 @@ htr_change_freq <- function(freq,
   # Create output folder if it doesn't exist
   htr_make_folder(outdir)
 
-  w <- parallel::detectCores() - 2
-
-  esms <- dir(indir, pattern = "*.nc", full.names = TRUE)
+  # Define workers
+  if(is.na(hpc)) {
+    w <- parallelly::availableCores(method = "system", omit = 2)
+  } else {
+    w <- parallelly::availableCores(method = "Slurm", omit = 2)
+  }
 
   ##############
 
@@ -50,6 +62,20 @@ htr_change_freq <- function(freq,
 
   ##############
 
+  if (hpc %in% c("array")) { # For hpc == "array", use the specific files as the starting point
+
+    esm <- dir(indir, pattern = file, full.names = TRUE)
+
+    if (stringr::str_to_lower(freq) == "yearly") { # run function
+      change_yearly(esm, outdir)
+    } else if (stringr::str_to_lower(freq) == "monthly") {
+      change_monthly(esm, outdir)
+    }
+
+  } else { # For hpc == "parallel" and non-hpc work, use the input directory as the starting point and run jobs in parallel
+
+  esms <- dir(indir, pattern = "*.nc", full.names = TRUE)
+
   future::plan(future::multisession, workers = w)
 
   if (stringr::str_to_lower(freq) == "yearly") {
@@ -59,4 +85,6 @@ htr_change_freq <- function(freq,
   }
 
   future::plan(future::sequential)
+
+  }
 }

R/htr_create_ensemble.R

@@ -1,15 +1,19 @@
 #' Create an ensemble based on list of models
 #'
+#'
 #' @inheritParams htr_slice_period
 #' @param model_list Character string of models to use for the ensemble
 #' @param variable The variable to create the ensemble for
 #' @param mean Use the mean (TRUE; default) or the median (FALSE) when creating the ensemble.
+#' @param season If using seasonal frequency, input the season name to detect the files
+#' @param domain If using depth-resolved models, input the domain name to detect the files
 #'
 #' @export
 #'
 #' @examples
 #' \dontrun{
 #' htr_create_ensemble(
+#' hpc = NA,
 #' indir = file.path(base_dir, "data", "proc", "regridded", "yearly", "tos"),
 #' outdir = file.path(base_dir, "data", "proc", "ensemble", "mean", "tos"),
 #' model_list = c("ACCESS-ESM1-5", "CanESM5"),
@@ -19,35 +23,76 @@
 #' mean = TRUE
 #' )
 #' }
-htr_create_ensemble <- function(indir,
+htr_create_ensemble <- function(hpc = NA, # if ran in the HPC, possible values are "array", "parallel"
+                                indir,
                                 outdir,
                                 model_list,
                                 variable = "tos",
                                 freq = "Omon",
                                 scenario = "historical",
+                                season = "", # default is no season
+                                domain = "", # default is no domain
                                 mean = TRUE # if false, use median
 ) {
 
   # Create output folder if it doesn't exist
   htr_make_folder(outdir)
 
-  w <- parallel::detectCores() - 2
+  # Define workers
+  if(is.na(hpc)) {
+    w <- parallelly::availableCores(methods = "system", omit = 2)
+  } else {
+    w <- parallelly::availableCores(methods = "Slurm", omit = 2)
+  }
+
+  ##############
+
+  if (stringr::str_length(domain) > 0 & stringr::str_length(season) > 0) {
+
+    files <- dir(indir, full.names = TRUE) %>%
+      stringr::str_subset(paste0("(?=.*", variable, "_", ")(?=.*", freq, "_", ")(?=.*", scenario, "_", ")(?=.*", season, "_", ")(?=.*", domain, ")")) %>%
+      stringr::str_subset(paste(model_list, collapse = "|"))
+
+  } else if (stringr::str_length(domain) > 0) {
+
+    files <- dir(indir, full.names = TRUE) %>%
+      stringr::str_subset(paste0("(?=.*", variable, "_", ")(?=.*", freq, "_", ")(?=.*", scenario, "_", ")(?=.*",  domain, ")")) %>%
+      stringr::str_subset(paste(model_list, collapse = "|"))
+
+  } else if (stringr::str_length(season) > 0) {
+
+    files <- dir(indir, full.names = TRUE) %>%
+      stringr::str_subset(paste0("(?=.*", variable, "_", ")(?=.*", freq, "_", ")(?=.*", scenario, "_", ")(?=.*",  season, ")")) %>%
+      stringr::str_subset(paste(model_list, collapse = "|"))
+
+  } else {
+
+    files <- dir(indir, full.names = TRUE) %>%
+      stringr::str_subset(paste0("(?=.*", variable, "_", ")(?=.*", freq, "_", ")(?=.*", scenario, ")")) %>%
+      stringr::str_subset(paste(model_list, collapse = "|"))
 
-  files <- dir(indir, full.names = TRUE) %>%
-    stringr::str_subset(paste0("(?=.*", variable, "_", ")(?=.*", freq, "_", ")(?=.*", scenario, "_", ")")) %>%
-    stringr::str_subset(paste(model_list, collapse = "|"))
+  }
 
   out_name <- files[1] %>%
     stringr::str_replace(indir, outdir) %>%
     stringr::str_replace(htr_get_CMIP6_bits(files[1])$Model, "ensemble")
 
+  ##############
+
   if (mean == TRUE) {
+
     cdo_code <- paste0("cdo -L -z zip -ensmean ", paste0(files, collapse = " "), " ", out_name)
+
   } else if (mean == FALSE) { # Calculate the median
+
     cdo_code <- paste0("cdo -L -z zip -ensmedian ", paste0(files, collapse = " "), " ", out_name)
+
   } else {
+
     print("Please provide the right option for mean")
+
   }
 
   system(cdo_code)
+
 }

R/htr_download_ESM.R

@@ -8,21 +8,26 @@
 #' @examples
 #' \dontrun{
 #' htr_download_ESM(
+#' hpc = NA,
 #' indir = file.path(base_dir, "data", "raw", "wget"), # input directory
 #' outdir = file.path(base_dir, "data", "raw", "tos") # output directory
 #' )
 #' }
-htr_download_ESM <- function(indir, # where wget files are located
+htr_download_ESM <- function(hpc = NA, # if ran in the HPC, possible values are "array", "parallel"
+                             indir, # where wget files are located
                              outdir) { # where .nc files should be downloaded
 
   # Create output folder if it doesn't exist
   htr_make_folder(outdir)
 
   pth <- getwd()
 
-  w <- parallel::detectCores() - 2
-
-  files <- dir(indir, pattern = "wget", full.names = TRUE)
+  # Define workers
+  if(is.na(hpc)) {
+    w <- parallelly::availableCores(methods = "system", omit = 2)
+  } else {
+    w <- parallelly::availableCores(methods = "Slurm", omit = 2)
+  }
 
   ##############
 
@@ -34,6 +39,8 @@ htr_download_ESM <- function(indir, # where wget files are located
 
   ##############
 
+  files <- dir(indir, pattern = "wget", full.names = TRUE)
+
   future::plan(future::multisession, workers = w)
   furrr::future_walk(files, wget_files)
   future::plan(future::sequential)

R/htr_fix_calendar.R

@@ -8,11 +8,28 @@
 #' @export
 #'
 #' @examples
-htr_fix_calendar <- function(indir) { # input directory
+#' \dontrun{
+#'
+#' htr_fix_calendar(
+#'   hpc = NA,
+#'   file = NA,
+#'   indir = file.path(base_dir, "data", "merged"), # input directory
+#' )
+#' }
+htr_fix_calendar <- function(hpc = NA, # if ran in the HPC, possible values are "array", "parallel"
+                             file = NA, # hpc = "array", the input will be the file
+                             indir) { # input directory
 
   . <- NULL # Stop devtools::check() complaints about NSE
 
-  w <- parallel::detectCores() - 2
+  # Define workers
+  if(is.na(hpc)) {
+    w <- parallelly::availableCores(method = "system", omit = 2)
+  } else {
+    w <- parallelly::availableCores(method = "Slurm", omit = 2)
+  }
+
+  ##############
 
   fix_cal <- function(f) {
     yrs <- ncdf4::nc_open(f) %>%
@@ -33,9 +50,22 @@ htr_fix_calendar <- function(indir) { # input directory
     }
   }
 
+  ##############
+
+  if (hpc %in% c("array")) { # For hpc == "array", use the specific files as the starting point
+
+    file <- dir(indir, pattern = file, full.names = TRUE)
+
+    fix_cal(file) # run function
+
+  } else { # For hpc == "parallel" and non-hpc work, use the input directory as the starting point and run jobs in parallel
+
+    netCDFs <- dir(indir, full.names = TRUE)
+
+    future::plan(future::multisession, workers = w)
+    furrr::future_walk(netCDFs, fix_cal) # run function in parallel
+    future::plan(future::sequential)
+
+  }
 
-  netCDFs <- dir(indir, full.names = TRUE)
-  future::plan(future::multisession, workers = w)
-  furrr::future_walk(netCDFs, fix_cal)
-  future::plan(future::sequential)
 }

R/htr_integrate_levels.R

@@ -0,0 +1,61 @@
+#' Get the weighted vertical means
+#'
+#' hjfjhfjhf
+#'
+#' @inheritParams htr_seasonal_frequency
+#' @param min_level Minimum level of depth domain
+#' @param max_level Maximum level of depth domain
+#' @param domain_name Depth domain name
+#'
+#' @return
+#' @export
+#'
+#' @examples
+#' \dontrun{
+#'
+#' }
+htr_integrate_levels <- function(indir,
+                                 tempdir,
+                                 outdir,
+                                 min_level,
+                                 max_level,
+                                 domain_name = ""
+) {
+
+  # Create output folder if it doesn't exist
+  htr_make_folder(outdir)
+
+  # Create temporary folder if it doesn't exist
+  htr_make_folder(tempdir)
+
+  w <- parallel::detectCores()-2
+
+  do_integrate <- function(f) {
+
+    if(stringr::str_length(domain_name) > 0) {
+      outname <- f %>%
+        basename() %>%
+        stringr::str_split("[.]") %>%
+        purrr::map(~paste0(.x[1], "_", domain_name, ".", .x[2]))
+    } else {
+      outname <- f %>%
+        basename()
+    }
+
+
+    out_file <- paste0(outdir, "/", outname)
+
+    system(paste0("cdo select,levrange=", min_level, ",", max_level, " ", f, " ", tempdir, "/", basename(f)))
+
+    system(paste0("cdo vertmean ", tempdir, "/", basename(f), " ", out_file))
+
+    print(basename(f))
+
+  }
+
+  esms <- dir(indir, pattern = "*.nc", full.names = TRUE)
+  future::plan(future::multisession, workers = w)
+  furrr::future_walk(esms, do_integrate)
+  future::plan(future::sequential)
+
+}

R/htr_merge_files.R

@@ -12,13 +12,15 @@
 #' \dontrun{
 #'
 #' htr_merge_files(
+#'   hpc = NA,
 #'   indir = file.path(base_dir, "data", "raw", "tos"), # input directory
 #'   outdir = file.path(base_dir, "data", "proc", "merged", "tos"), # output directory
 #'   year_start = 1985, # earliest year across all the scenarios considered
 #'   year_end = 2100 # latest year across all the scenarios considered
 #' )
 #' }
-htr_merge_files <- function(indir, # where nc files are located
+htr_merge_files <- function(hpc = NA, # if ran in the HPC, possible values are "array", "parallel"
+                            indir, # where nc files are located
                             outdir, # where merged files should be saved
                             year_start, # start year of historical file
                             year_end # end year of projection file
@@ -28,7 +30,12 @@ htr_merge_files <- function(indir, # where nc files are located
   # Create output folder if it doesn't exist
   htr_make_folder(outdir)
 
-  w <- parallel::detectCores() - 2
+  # Define workers
+  if(is.na(hpc)) {
+    w <- parallelly::availableCores(methods = "system", omit = 2)
+  } else {
+    w <- parallelly::availableCores(methods = "Slurm", omit = 2)
+  }
 
   l <- htr_get_meta(indir, string = c("Variable", "Frequency", "Scenario", "Model", "Variant"))