Merge pull request #4083 from sarahkanee/fix/chirps-preprocess-dates

Set up config file for irrigation workflow

Author: mdietze

modules/data.remote/inst/ccmmf/documentation/setup_env.sh

@@ -11,8 +11,6 @@
 #   products/inventory                         -- Management Tracking outputs
 # Finished tree: documentation/sessions/00-setup.md (Data layout).
 # Product overview: ../README.md. Paths created in Session 0.
-# Irrigation: workflows/irrigation-statewide/config_paths.yml keys should point into this tree;
-#   CHIRPS_DIR/CIMIS_DIR are raw staging; parcel extracts from preprocessing/ may differ.
 
 # Workspace roots. Set CCMMF_BASE once; CODE and ROOT follow unless you override them.
 export CCMMF_BASE="${CCMMF_BASE:-$HOME}"
@@ -55,11 +53,11 @@ export HLS_CREDENTIAL_FOLDER="${HLS_CREDENTIAL_FOLDER:-$HOME}"            # dir
 export CDL_DIR="${CDL_DIR:-$CCMMF_ROOT/CDL}"                              # CDL GeoTIFF
 
 export CLIMATE_ROOT="${CLIMATE_ROOT:-$CCMMF_ROOT/climate}"
-export CHIRPS_DIR="${CHIRPS_DIR:-$CLIMATE_ROOT/CHIRPS}"   # raw CHIRPS staging; irrig YAML chirps_precip_path = preprocess extract dir
-export CIMIS_DIR="${CIMIS_DIR:-$CLIMATE_ROOT/CIMIS}"       # raw CIMIS staging; irrig YAML cimis_etref_path = preprocess extract dir
+export CHIRPS_DIR="${CHIRPS_DIR:-$CLIMATE_ROOT/CHIRPS}"   # raw CHIRPS staging
+export CIMIS_DIR="${CIMIS_DIR:-$CLIMATE_ROOT/CIMIS}"       # raw CIMIS staging
 
 export SOILS_ROOT="${SOILS_ROOT:-$CCMMF_ROOT/soils}"
-export SSURGO_DIR="${SSURGO_DIR:-$SOILS_ROOT/SSURGO}"      # gdb + weights; also set irrig ssurgo_* YAML keys
+export SSURGO_DIR="${SSURGO_DIR:-$SOILS_ROOT/SSURGO}"      # gdb + weights
 
 # --- Lookups ---
 export LOOKUPS_ROOT="${LOOKUPS_ROOT:-$CCMMF_ROOT/lookups}"
@@ -85,6 +83,31 @@ export COUNTY_TRANSITION_MATRICES_DIR="${COUNTY_TRANSITION_MATRICES_DIR:-$LANDIQ
 # Scheduler-agnostic submit (sbatch, qsub, or local). See documentation/submit_job.sh
 export CCMMF_SUBMIT="${CCMMF_SUBMIT:-$CCMMF_CODE/documentation/submit_job.sh}"
 
+export CHIRPS_PRECIP_PATH="${CHIRPS_PRECIP_PATH:-$PRODUCTS_INVENTORY/irrigation/chirps-extracted}"
+export CIMIS_ETREF_PATH="${CIMIS_ETREF_PATH:-$PRODUCTS_INVENTORY/irrigation/cimis-extracted}"
+export CHIRPS_PREPROCESS_DIR="${CHIRPS_PREPROCESS_DIR:-_results_chirps}"
+export CIMIS_PREPROCESS_DIR="${CIMIS_PREPROCESS_DIR:-_results_v2}"
+export SSURGO_PREPROCESS_DIR="${SSURGO_PREPROCESS_DIR:-_results}"
+
+cat > "$(dirname "$IRRIG_PREPROCESS")/config_paths.yml" <<EOF
+default:
+  landiq_parcels_gpkg: "$LANDIQ_HARMONIZED/parcels-consolidated.gpkg"
+  year1: $PRIOR_YEAR
+  year2: $TARGET_YEAR
+  chirps_dir: "$CHIRPS_DIR"
+  chirps_preprocess_dir: "$CHIRPS_PREPROCESS_DIR"
+  chirps_precip_path: "$CHIRPS_PRECIP_PATH"
+  cimis_dir: "$CIMIS_DIR"
+  cimis_preprocess_dir: "$CIMIS_PREPROCESS_DIR"
+  cimis_etref_path: "$CIMIS_ETREF_PATH"
+  ssurgo_gdb_path: "$SSURGO_DIR/gSSURGO_CA.gdb"
+  ssurgo_preprocess_dir: "$SSURGO_PREPROCESS_DIR"
+  ssurgo_weights_path: "$SSURGO_DIR/ssurgo-weights.parquet"
+  event_output_dir: "$EVENT_OUTPUT_DIR"
+  crops_path: "$LANDIQ_GAPFILLED/crops_all_years.parq"
+  mslsp_path: "$MATCHED_DIR"
+EOF
+
 echo "[setup_env] BASE=$CCMMF_BASE ROOT=$CCMMF_ROOT CODE=$CCMMF_CODE YEARS=$PRIOR_YEAR/$TARGET_YEAR"
 echo "[setup_env] LANDIQ_GAPFILLED=$LANDIQ_GAPFILLED"
 echo "[setup_env] HLS_ROOT=$HLS_ROOT HLS_PARCEL_TILES_DIR=$HLS_PARCEL_TILES_DIR"

workflows/irrigation-statewide/_targets.R

@@ -19,7 +19,7 @@ config_base <- config::get(
 )
 config_paths <- config::get(
   file = file.path(root_dir, "config_paths.yml"),
-  config = Sys.getenv("IRRIGATION_PATHS_CONFIG", "default")
+  config = "default"
 )
 config <- config::merge(config_base, config_paths)
 

workflows/irrigation-statewide/config_paths.yml

@@ -1,8 +1,18 @@
 default:
+  # Preprocess
+  landiq_parcels_gpkg: "/projectnb/dietzelab/ccmmf/LandIQ-harmonized-v4.1/parcels.gpkg"
+  year1: 2023
+  year2: 2024
+  chirps_dir: "/projectnb/dietzelab/ccmmf/management/irrigation/"
+  chirps_preprocess_dir: "_results_chirps"
+  chirps_precip_path: "/projectnb/dietzelab/ccmmf/data/chirps-extracted"
+  cimis_dir: "/projectnb/dietzelab/ccmmf/data_raw/cimis/cimis"
+  cimis_preprocess_dir: "_results_v2"
+  cimis_etref_path: "/projectnb/dietzelab/ccmmf/data/cimis-extracted"
+  ssurgo_gdb_path: "/projectnb/dietzelab/ccmmf/data_raw/ssurgo/gSSURGO_CA.gdb"
+  ssurgo_preprocess_dir: "_results"
+  ssurgo_weights_path: "/projectnb/dietzelab/ccmmf/data_raw/ssurgo/ssurgo-weights.parquet"
+  # tar_make
   event_output_dir: "/projectnb/dietzelab/ccmmf/usr/ashiklom/event-outputs"
   crops_path: "/projectnb/dietzelab/ccmmf/LandIQ-harmonized-v4.1/crops_all_years.parq"
   mslsp_path: "/projectnb/dietzelab/ccmmf/management/phenology/matched_landiq_mslsp_v4.1"
-  cimis_etref_path: "/projectnb/dietzelab/ccmmf/data/cimis-extracted"
-  chirps_precip_path: "/projectnb/dietzelab/ccmmf/data/chirps-extracted"
-  ssurgo_weights_path: "/projectnb/dietzelab/ccmmf/data_raw/ssurgo/ssurgo-weights.parquet"
-  ssurgo_gdb_path: "/projectnb/dietzelab/ccmmf/data_raw/ssurgo/gSSURGO_CA.gdb"

workflows/irrigation-statewide/preprocessing/chirps-preprocess.R

@@ -1,18 +1,20 @@
 #!/usr/bin/env Rscript
 
-chirps_dir <- Sys.getenv("CHIRPS_DIR", "/projectnb/dietzelab/ccmmf/management/irrigation/")
-chirps_outdir <- Sys.getenv("CHIRPS_EXTRACT_DIR", "_results_chirps")
+root_dir <- here::here("workflows/irrigation-statewide")
+cfg <- config::get(
+  file = file.path(root_dir, "config_paths.yml"),
+  config = "default"
+)
+
+chirps_dir <- cfg[["chirps_dir"]]
+chirps_outdir <- cfg[["chirps_preprocess_dir"]]
+parcel_file <- cfg[["landiq_parcels_gpkg"]]
 chirpsfiles <- list.files(
   chirps_dir,
   "chirps-v2.0.*.nc",
   full.names = TRUE
 )
 
-parcel_file <- Sys.getenv(
-  "LANDIQ_PARCELS_GPKG",
-  "/projectnb/dietzelab/ccmmf/LandIQ-harmonized-v4.1/parcels.gpkg"
-)
-
 extract_chirps <- function(fname, parcel_file, outdir = "_results_chirps") {
   # fname <- chirpsfiles[[1]]
   parcels_sf <- sf::read_sf(parcel_file, use_stream = TRUE)
@@ -52,12 +54,12 @@ extract_chirps <- function(fname, parcel_file, outdir = "_results_chirps") {
     tidyr::pivot_longer(
       -c("parcel_id"),
       names_to = "yday",
-      names_pattern = ".*\\.days_p05_(\\d+)$",
+      names_pattern = ".*_(\\d+)$",
       names_transform = as.integer,
       values_to = "precip_mm_day"
     ) |>
     dplyr::mutate(
-      date = date0 + .data$yday,
+      date = date0 + .data$yday - 1L,  # precip_1: Jan 1 + 1 = Jan 2 without -1
       .keep = "unused"
     ) |>
     dplyr::relocate("date", .after = "parcel_id") |>

workflows/irrigation-statewide/preprocessing/cimis-01-weights.R

@@ -7,23 +7,27 @@ library(Matrix)
 library(exactextractr)
 library(arrow)
 
-parcel_file <- Sys.getenv(
-  "LANDIQ_PARCELS_GPKG",
-  "/projectnb/dietzelab/ccmmf/LandIQ-harmonized-v4.1/parcels.gpkg"
+root_dir <- here::here("workflows/irrigation-statewide")
+cfg <- config::get(
+  file = file.path(root_dir, "config_paths.yml"),
+  config = "default"
 )
-cimis_dir <- Sys.getenv("CIMIS_DIR", "/projectnb/dietzelab/ccmmf/data_raw/cimis/cimis")
 
-outdir <- "_results_v2"
+parcel_file <- cfg[["landiq_parcels_gpkg"]]
+cimis_dir <- cfg[["cimis_dir"]]
+year1 <- as.character(cfg[["year1"]])
+year2 <- as.character(cfg[["year2"]])
+outdir <- cfg[["cimis_preprocess_dir"]]
 dir.create(outdir, showWarnings = FALSE, recursive = TRUE)
 
 parcels_sf <- read_sf(parcel_file, use_stream = TRUE)
 
-cimis_f0 <- file.path(cimis_dir, "2019", "01", "01", "ETo.asc.gz")
+cimis_f0 <- file.path(cimis_dir, year1, "01", "01", "ETo.asc.gz")
 stopifnot(file.exists(cimis_f0))
 
 # A subset of CIMIS files, like this one, have slightly smaller dimensions for
 # some reason. Compute their weights separately.
-cimis_falt <- file.path(cimis_dir, "2015", "03", "01", "ETo.asc.gz")
+cimis_falt <- file.path(cimis_dir, year2, "01", "01", "ETo.asc.gz")
 stopifnot(file.exists(cimis_falt))
 
 n_parcels <- nrow(parcels_sf)

workflows/irrigation-statewide/preprocessing/cimis-02-extract.R

@@ -13,21 +13,25 @@ library(purrr)
 library(arrow)
 library(clustermq)
 
+root_dir <- here::here("workflows/irrigation-statewide")
+cfg <- config::get(
+  file = file.path(root_dir, "config_paths.yml"),
+  config = "default"
+)
+
 n_workers <- as.integer(Sys.getenv("CLUSTERMQ_N_JOBS", "20"))
 walltime <- "02:00:00"
 
-cimis_root <- Sys.getenv("CIMIS_DIR", "/projectnb/dietzelab/ccmmf/data_raw/cimis/cimis")
-
-outdir <- "_results_v2/daily-raw"
+cimis_root <- cfg[["cimis_dir"]]
+preprocess_dir <- cfg[["cimis_preprocess_dir"]]
+outdir <- file.path(preprocess_dir, "daily-raw")
 dir.create(outdir, recursive = TRUE, showWarnings = FALSE)
 
-W <- readRDS("_results_v2/spatial_weights.rds")
+W <- readRDS(file.path(preprocess_dir, "spatial_weights.rds"))
+weights_alt <- file.path(preprocess_dir, "spatial_weights_alt.rds")
 
 cimis_manifest <- "cimis_files.txt"
-years <- seq(
-  as.integer(Sys.getenv("START_YEAR", "2015")),
-  as.integer(Sys.getenv("END_YEAR", "2024"))
-)
+years <- seq(as.integer(cfg[["year1"]]), as.integer(cfg[["year2"]]))
 if (!file.exists(cimis_manifest)) {
   get_cimis_files <- function(year) {
     ydir <- file.path(cimis_root, year)
@@ -47,7 +51,7 @@ if (!file.exists(cimis_manifest)) {
 }
 
 # Extract
-process_file <- function(fname, W, outdir) {
+process_file <- function(fname, W, outdir, weights_alt) {
   day <- basename(dirname(fname))
   month <- basename(dirname(dirname(fname)))
   year <- basename(dirname(dirname(dirname(fname))))
@@ -66,7 +70,7 @@ process_file <- function(fname, W, outdir) {
   rsize <- terra::size(r)
   if (rsize == 285600) {
     # Alternate size files -- use alternate weights
-    W <- readRDS("_results_v2/spatial_weights_alt.rds")
+    W <- readRDS(weights_alt)
   } else if (rsize != 276000) {
     stop("File ", fname, " has unexpected size ", rsize)
   }
@@ -99,8 +103,8 @@ process_file <- function(fname, W, outdir) {
 cimis_long <- Q(
   fun     = process_file,
   fname = cimis_files,
-  const   = list(W = W, outdir = outdir),
-  n_jobs  = n_workers,         # SGE array size -- persistent worker processes
+  const   = list(W = W, outdir = outdir, weights_alt = weights_alt),
+  n_jobs  = n_workers,         # SGE array size — persistent worker processes
   template = list(cores = 1, walltime = walltime),
   fail_on_error = FALSE
 )

workflows/irrigation-statewide/preprocessing/ssurgo-01-spatial-weights.R

@@ -12,7 +12,13 @@ library(dplyr)
 library(arrow)  # for efficient weight storage
 library(clustermq)
 
-outdir <- "_results"
+root_dir <- here::here("workflows/irrigation-statewide")
+cfg <- config::get(
+  file = file.path(root_dir, "config_paths.yml"),
+  config = "default"
+)
+
+outdir <- cfg[["ssurgo_preprocess_dir"]]
 dir.create(outdir, showWarnings = FALSE, recursive = TRUE)
 
 # Number of parcels to process at a time
@@ -21,10 +27,7 @@ dir.create(outdir, showWarnings = FALSE, recursive = TRUE)
 parcel_chunk_size <- 2000
 
 # Transform ssurgo mupolygon column to parquet for faster reading
-gdb_path <- Sys.getenv(
-  "SSURGO_GDB_PATH",
-  "/projectnb/dietzelab/ccmmf/data_raw/ssurgo/gSSURGO_CA.gdb"
-)
+gdb_path <- cfg[["ssurgo_gdb_path"]]
 mupoly_path <- "./ssurgo_mupolygons.parquet"
 if (!file.exists(mupoly_path)) {
   message("Creating parquet version of mupolygons for faster reads")
@@ -44,10 +47,7 @@ if (!file.exists(mupoly_path)) {
   rm(mupoly_raw, mupoly)
 }
 
-parcels_path <- Sys.getenv(
-  "LANDIQ_PARCELS_GPKG",
-  "/projectnb/dietzelab/ccmmf/LandIQ-harmonized-v4.1/parcels.gpkg"
-) #nolint
+parcels_path <- cfg[["landiq_parcels_gpkg"]]
 
 parcel_ids <- st_read(
   parcels_path,

workflows/irrigation-statewide/preprocessing/ssurgo-02-combine.R

@@ -3,25 +3,33 @@
 library(arrow)
 library(duckdb)
 
+root_dir <- here::here("workflows/irrigation-statewide")
+cfg <- config::get(
+  file = file.path(root_dir, "config_paths.yml"),
+  config = "default"
+)
+
+preprocess_dir <- cfg[["ssurgo_preprocess_dir"]]
+weights_out <- cfg[["ssurgo_weights_path"]]
+
 conn <- dbConnect(duckdb(), dbdir = ":memory:")
 
 dbExecute(conn, "SET memory_limit = '32GB'")
 
-dbExecute(conn, "
+dbExecute(conn, sprintf("
   COPY (
     SELECT
     *
-    FROM read_parquet('_results/*.parquet')
+    FROM read_parquet('%s/*.parquet')
     ORDER BY parcel_id ASC
   )
-  TO 'ssurgo-weights.parquet'
+  TO '%s'
   (
     FORMAT PARQUET,
     OVERWRITE_OR_IGNORE,
     COMPRESSION 'ZSTD'
   );
-  "
-)
+  ", preprocess_dir, weights_out))
 
 dbDisconnect(conn, shutdown = TRUE)
 
@@ -30,7 +38,7 @@ message(
   "Testing to confirm we can open the data ",
   "and it produces valid weights."
 )
-dat <- open_dataset("ssurgo-weights.parquet")
+dat <- open_dataset(weights_out)
 
 dsub <- dat |>
   dplyr::filter(parcel_id %in% c(1, 100, 1000, 10000, 100000)) |>