make workflow configs parseable outside R and resolve input paths from env

workflows/fertilization-statewide/01-build-parcel-design.R

@@ -3,9 +3,54 @@
 config <- config::get(file = "workflows/fertilization-statewide/config.yml",
                       config = Sys.getenv("FERT_PROJECT", "default"))
 
+# paths in config.yml are relative to ccmmf_dir so the yaml stays plain data any
+# parser can read. a value from the environment wins and is used as given, so a
+# run can point anywhere without editing the file
+ccmmf_dir <- Sys.getenv("CCMMF_DIR")
+if (!nzchar(ccmmf_dir)) {
+  ccmmf_dir <- config[["ccmmf_dir"]]
+}
+# an override set to an empty string is treated as unset, otherwise file.path
+# would build a path rooted at /
+resolve_path <- function(key, env_var) {
+  p <- Sys.getenv(env_var)
+  if (!nzchar(p)) {
+    p <- file.path(ccmmf_dir, config[[key]])
+  }
+  path.expand(p)
+}
+crops_path      <- resolve_path("crops_path", "CCMMF_CROPS_PATH")
+phen_dir        <- resolve_path("phen_dir", "CCMMF_PHEN_DIR")
+pft_lookup_path <- resolve_path("pft_lookup_path", "CCMMF_PFT_LOOKUP")
+output_dir      <- resolve_path("output_dir", "CCMMF_FERT_OUT")
+phen_glob <- Sys.getenv("CCMMF_PHEN_GLOB")
+if (!nzchar(phen_glob)) {
+  phen_glob <- config[["phen_glob"]]
+}
+
+# inputs are required, not optional. away from SCC the defaults will not
+# resolve, so name the variable to set rather than failing deeper in a read
+inputs <- c(CCMMF_CROPS_PATH = crops_path, CCMMF_PHEN_DIR = phen_dir,
+            CCMMF_PFT_LOOKUP = pft_lookup_path)
+if (!all(file.exists(inputs))) {
+  absent <- inputs[!file.exists(inputs)]
+  PEcAn.logger::logger.severe(
+    "input not found: ", paste(absent, collapse = ", "),
+    ". Set ", paste(names(absent), collapse = ", "), " or CCMMF_DIR.")
+}
+
+# log what resolved so a run can be reconstructed from its output alone
+PEcAn.logger::logger.info(paste0(
+  "\nResolved paths\n",
+  "  crops_path      : ", crops_path, "\n",
+  "  phen_dir        : ", phen_dir, "\n",
+  "  phen_glob       : ", phen_glob, "\n",
+  "  pft_lookup_path : ", pft_lookup_path, "\n",
+  "  output_dir      : ", output_dir, "\n"), wrap = FALSE)
+
 set.seed(config[["seed"]])
 
-staging_dir <- file.path(config[["output_dir"]], "_staging")
+staging_dir <- file.path(output_dir, "_staging")
 dir.create(staging_dir, showWarnings = FALSE, recursive = TRUE)
 
 options(arrow.unsafe_metadata = TRUE)
@@ -111,7 +156,7 @@ crops <- DBI::dbGetQuery(con, sprintf(
           CAST(season AS INTEGER) AS season, CLASS,
           TRY_CAST(NULLIF(NULLIF(TRIM(CAST(SUBCLASS AS VARCHAR)), '**'), '') AS INTEGER) AS SUBCLASS
    FROM read_parquet('%s') WHERE \"year\" IN (%s) AND CLASS IS NOT NULL",
-  config[["crops_path"]], yr_list)) |>
+  crops_path, yr_list)) |>
   dplyr::rename(year = "yr") |>
   dplyr::mutate(code = paste0(.data$CLASS, .data$SUBCLASS))
 
@@ -120,7 +165,7 @@ anchor_cols <- sort(unique(pft_anchor))
 
 # "**" is the LandIQ sentinel for subclass not specified. it becomes NA on both
 # sides of the join, so those rows land on the class-level fallback
-pft_lookup <- readr::read_csv(config[["pft_lookup_path"]],
+pft_lookup <- readr::read_csv(pft_lookup_path,
                               show_col_types = FALSE) |>
   dplyr::filter(!is.na(.data$PFT))
 pft_by_code <- pft_lookup |>
@@ -138,7 +183,7 @@ phen <- DBI::dbGetQuery(con, sprintf(
           CAST(season AS INTEGER) AS season, gapfill_date_source, %s
    FROM read_parquet('%s') WHERE \"year\" IN (%s)",
   paste(anchor_cols, collapse = ", "),
-  file.path(config[["phen_dir"]], config[["phen_glob"]]), yr_list))
+  file.path(phen_dir, phen_glob), yr_list))
 
 missing_cols <- setdiff(anchor_cols, names(phen))
 if (length(missing_cols) > 0) {

workflows/fertilization-statewide/02-sample-n-rates.R

@@ -3,9 +3,20 @@
 config <- config::get(file = "workflows/fertilization-statewide/config.yml",
                       config = Sys.getenv("FERT_PROJECT", "default"))
 
+# an override set to an empty string is treated as unset
+ccmmf_dir <- Sys.getenv("CCMMF_DIR")
+if (!nzchar(ccmmf_dir)) {
+  ccmmf_dir <- config[["ccmmf_dir"]]
+}
+output_dir <- Sys.getenv("CCMMF_FERT_OUT")
+if (!nzchar(output_dir)) {
+  output_dir <- file.path(ccmmf_dir, config[["output_dir"]])
+}
+output_dir <- path.expand(output_dir)
+
 set.seed(config[["seed"]])
 
-staging_dir <- file.path(config[["output_dir"]], "_staging")
+staging_dir <- file.path(output_dir, "_staging")
 design_file <- file.path(staging_dir, "_staging_01_design.rds")
 if (!file.exists(design_file)) {
   PEcAn.logger::logger.severe(

workflows/fertilization-statewide/03-write-parquet.R

@@ -3,7 +3,18 @@
 config <- config::get(file = "workflows/fertilization-statewide/config.yml",
                       config = Sys.getenv("FERT_PROJECT", "default"))
 
-staging_dir <- file.path(config[["output_dir"]], "_staging")
+# override set to an empty string is treated as unset
+ccmmf_dir <- Sys.getenv("CCMMF_DIR")
+if (!nzchar(ccmmf_dir)) {
+  ccmmf_dir <- config[["ccmmf_dir"]]
+}
+output_dir <- Sys.getenv("CCMMF_FERT_OUT")
+if (!nzchar(output_dir)) {
+  output_dir <- file.path(ccmmf_dir, config[["output_dir"]])
+}
+output_dir <- path.expand(output_dir)
+
+staging_dir <- file.path(output_dir, "_staging")
 events_file <- file.path(staging_dir, "_staging_02_events.rds")
 if (!file.exists(events_file)) {
   PEcAn.logger::logger.severe(
@@ -40,7 +51,7 @@ out <- events |>
     crop_code = .data$code
   )
 
-out_path <- config[["output_dir"]]
+out_path <- output_dir
 dir.create(out_path, showWarnings = FALSE, recursive = TRUE)
 
 ## clean prior shards
@@ -86,8 +97,19 @@ if (workers > 1) {
 
 # mclapply does not raise when a worker fails: it returns a try-error for an R
 # level error and NULL for a killed worker, which is the realistic out of memory
-# case. write_batch returns the shard path, so anything that is not a path failed
-failed <- !vapply(written, is.character, logical(1))
+# case. write_batch returns the shard path, so anything that is not a path failed.
+# a try-error is itself a character vector, so is.character alone lets it through
+failed <- vapply(
+  written,
+  function(x) {
+    is.null(x) ||
+      inherits(x, "try-error") ||
+      !is.character(x) ||
+      length(x) != 1L ||
+      !file.exists(x)
+  },
+  logical(1)
+)
 if (any(failed)) {
   PEcAn.logger::logger.severe(sprintf(
     "%d of %d shard writes failed", sum(failed), length(written)))

workflows/fertilization-statewide/README.md

@@ -8,8 +8,9 @@ Source: California Department of Water Resources. (2016-2023). Statewide Crop Ma
 
 Configuration parameters live in `config.yml`. Most setups only need:
 
-- `crops_path`: the harmonized CADWR Land Use crops parquet. Override with `CCMMF_CROPS_PATH`
-- `phen_dir`: the gap-filled LandIQ to MSLSP match directory, the same product the ncc workflow reads. Override with `CCMMF_PHEN_DIR`
+- `ccmmf_dir`: data root the input paths are relative to. Override with `CCMMF_DIR`
+- `crops_path`: gap-filled LandIQ crops table. Override with `CCMMF_CROPS_PATH`
+- `phen_dir`: gap-filled LandIQ to MSLSP match. Override with `CCMMF_PHEN_DIR`
 - `phen_glob`: file glob under `phen_dir` (default `assigned_year=*_gapfilled.parquet`). Override with `CCMMF_PHEN_GLOB`
 - `pft_lookup_path`: crop code to PFT table, the same one the monitoring products use. Override with `CCMMF_PFT_LOOKUP`
 - `pft_anchor`: phenology transition used as the anchor, per PFT. See Application timing
@@ -68,6 +69,42 @@ schedule.
 - Only crop codes present in the crosswalk resolve to an N rate envelope. Cycles whose
   code does not resolve are dropped and reported at run time.
 
+# Running outside the BU cluster
+
+`config.yml` is plain YAML with no R expressions, so any parser can read it. Input paths
+are relative to `ccmmf_dir` and are resolved at run time; `01` logs every resolved path,
+so a run's log records the configuration it actually used.
+
+Three inputs are needed. They are monitoring products, not distributed with this repo:
+
+| config key | what it is |
+|---|---|
+| `crops_path` | gap-filled LandIQ crops table, one row per parcel, year and season |
+| `phen_dir` | gap-filled LandIQ to MSLSP phenology match, keyed the same way |
+| `pft_lookup_path` | crop code to PFT table |
+
+If they sit under one directory in the layout `config.yml` describes, point `CCMMF_DIR` at
+it and nothing else changes:
+
+```
+CCMMF_DIR=/my/ccmmf FERT_PROJECT=all bash workflows/fertilization-statewide/run-statewide.sh
+```
+
+If the layout differs, override paths individually. These take precedence over
+`ccmmf_dir`:
+
+```
+CCMMF_CROPS_PATH=/data/crops_all_years.parq \
+CCMMF_PHEN_DIR=/data/phenology \
+CCMMF_PFT_LOOKUP=/data/LandIQ_cropCode_lookup_table.csv \
+CCMMF_FERT_OUT=/data/events \
+FERT_PROJECT=all bash workflows/fertilization-statewide/run-statewide.sh
+```
+
+`phen_glob` can be overridden with `CCMMF_PHEN_GLOB` if the phenology files are named
+differently. Inputs are required rather than optional: if one does not resolve, the run
+stops and names the variable to set.
+
 # Output
 
 Parcel range sharded parquet at `<output_dir>/`. Columns: `parcel_id`, `ens_id` (`ens_NNN`, shared with ncc workflow), `date`, `nh4_n_kg_m2`, `no3_n_kg_m2`, `org_c_kg_m2` (zero), `org_n_kg_m2` (zero), `crop_code`.

workflows/fertilization-statewide/check-result.R

@@ -3,9 +3,20 @@
 config <- config::get(file = "workflows/fertilization-statewide/config.yml",
                       config = Sys.getenv("FERT_PROJECT", "default"))
 
+# an override set to an empty string is treated as unset
+ccmmf_dir <- Sys.getenv("CCMMF_DIR")
+if (!nzchar(ccmmf_dir)) {
+  ccmmf_dir <- config[["ccmmf_dir"]]
+}
+output_dir <- Sys.getenv("CCMMF_FERT_OUT")
+if (!nzchar(output_dir)) {
+  output_dir <- file.path(ccmmf_dir, config[["output_dir"]])
+}
+output_dir <- path.expand(output_dir)
+
 options(arrow.unsafe_metadata = TRUE)
 
-out_path <- config[["output_dir"]]
+out_path <- output_dir
 if (!dir.exists(out_path)) {
   PEcAn.logger::logger.severe("Output not found: ", out_path)
 }

workflows/fertilization-statewide/config.yml

@@ -8,19 +8,22 @@ default:
   nh4_fraction: 1
   seed: 42
   years: [2016, 2018, 2019, 2020, 2021, 2022, 2023]
-  # crops is the gap-filled LandIQ table (v4.1.2), the same inventory the
-  # phenology match is built against. v4.1 is the ungapfilled table and covers
-  # parcels the match does not.
-  # large inputs are not distributed with the repo. the defaults resolve on SCC;
-  # set the matching environment variable to run anywhere else. final home for
-  # these is pending ccmmf/organization#257
-  crops_path: !expr path.expand(Sys.getenv("CCMMF_CROPS_PATH", "/projectnb/dietzelab/ccmmf/LandIQ-harmonized-v4.1.2/crops_all_years.parq"))
+  # paths below are relative to ccmmf_dir. so any yaml parser can read this file;
+  # the scripts resolve them. the defaults resolve on SCC. off it, set CCMMF_DIR 
+  # to the data root, or override any one path with its own variable. final home 
+  # for these is pending ccmmf/organization#257
+  ccmmf_dir: /projectnb/dietzelab/ccmmf
+
+  # gap-filled LandIQ table, the same inventory the phenology match is built
+  # against. the ungapfilled table covers parcels the match does not
+  crops_path: LandIQ-harmonized-v4.1.2/crops_all_years.parq
+
   # gap-filled LandIQ to MSLSP match, the same product the ncc workflow reads
-  phen_dir: !expr path.expand(Sys.getenv("CCMMF_PHEN_DIR", "/projectnb/dietzelab/ccmmf/management/phenology/matched_landiq_mslsp_v4.1.2/gapfill_dates"))
-  phen_glob: !expr Sys.getenv("CCMMF_PHEN_GLOB", "assigned_year=*_gapfilled.parquet")
+  phen_dir: management/phenology/matched_landiq_mslsp_v4.1.2/gapfill_dates
+  phen_glob: "assigned_year=*_gapfilled.parquet"
 
-  # crop code to PFT. same table the monitoring products use
-  pft_lookup_path: !expr path.expand(Sys.getenv("CCMMF_PFT_LOOKUP", "/projectnb/dietzelab/ccmmf/management/LandIQ_cropCode_lookup_table.csv"))
+  # crop code to PFT
+  pft_lookup_path: management/LandIQ_cropCode_lookup_table.csv
 
   # anchor transition per PFT. annuals are timed to planting and perennials to
   # leaf-on, matching the annual/perennial split the monitoring event products
@@ -34,7 +37,7 @@ default:
   crosswalk_path: workflows/fertilization-statewide/crop_name_crosswalk.tsv
   # versioned alongside ncc: v2.0 is the pass built on the v4.1.2 monitoring
   # products. bump rather than writing over
-  output_dir: !expr path.expand(Sys.getenv("CCMMF_FERT_OUT", "/projectnb/dietzelab/ccmmf/usr/akash/event_files/fertilization/v2.0"))
+  output_dir: usr/akash/event_files/fertilization/v2.0
   batch_size: 100
   workers: 1
 

workflows/ncc-statewide/01-build-parcel-design.R

@@ -3,9 +3,54 @@
 config <- config::get(file = "workflows/ncc-statewide/config.yml",
                       config = Sys.getenv("NCC_PROJECT", "default"))
 
+# paths in config.yml are relative to ccmmf_dir so the yaml stays plain data any
+# parser can read. a value from the environment wins and is used as given, so a
+# run can point anywhere without editing the file
+ccmmf_dir <- Sys.getenv("CCMMF_DIR")
+if (!nzchar(ccmmf_dir)) {
+  ccmmf_dir <- config[["ccmmf_dir"]]
+}
+# an override set to an empty string is treated as unset, otherwise file.path
+# would build a path rooted at /
+resolve_path <- function(key, env_var) {
+  p <- Sys.getenv(env_var)
+  if (!nzchar(p)) {
+    p <- file.path(ccmmf_dir, config[[key]])
+  }
+  path.expand(p)
+}
+crops_path      <- resolve_path("crops_path", "CCMMF_CROPS_PATH")
+phen_dir        <- resolve_path("phen_dir", "CCMMF_PHEN_DIR")
+pft_lookup_path <- resolve_path("pft_lookup_path", "CCMMF_PFT_LOOKUP")
+output_dir      <- resolve_path("output_dir", "CCMMF_NCC_OUT")
+phen_glob <- Sys.getenv("CCMMF_PHEN_GLOB")
+if (!nzchar(phen_glob)) {
+  phen_glob <- config[["phen_glob"]]
+}
+
+# inputs are required, not optional. away from SCC the defaults will not
+# resolve, so name the variable to set rather than failing deeper in a read
+inputs <- c(CCMMF_CROPS_PATH = crops_path, CCMMF_PHEN_DIR = phen_dir,
+            CCMMF_PFT_LOOKUP = pft_lookup_path)
+if (!all(file.exists(inputs))) {
+  absent <- inputs[!file.exists(inputs)]
+  PEcAn.logger::logger.severe(
+    "input not found: ", paste(absent, collapse = ", "),
+    ". Set ", paste(names(absent), collapse = ", "), " or CCMMF_DIR.")
+}
+
+# log what resolved so a run can be reconstructed from its output alone
+PEcAn.logger::logger.info(paste0(
+  "\nResolved paths\n",
+  "  crops_path      : ", crops_path, "\n",
+  "  phen_dir        : ", phen_dir, "\n",
+  "  phen_glob       : ", phen_glob, "\n",
+  "  pft_lookup_path : ", pft_lookup_path, "\n",
+  "  output_dir      : ", output_dir, "\n"), wrap = FALSE)
+
 set.seed(config[["seed"]])
 
-staging_dir <- file.path(config[["output_dir"]], "_staging")
+staging_dir <- file.path(output_dir, "_staging")
 dir.create(staging_dir, showWarnings = FALSE, recursive = TRUE)
 
 options(arrow.unsafe_metadata = TRUE)
@@ -31,7 +76,7 @@ if (any(timing$offset_min > timing$offset_max)) {
 
 # "**" is the LandIQ sentinel for subclass not specified. it becomes NA on both
 # sides of the join, so those rows land on the class-level fallback
-pft_lookup <- readr::read_csv(config[["pft_lookup_path"]],
+pft_lookup <- readr::read_csv(pft_lookup_path,
                               show_col_types = FALSE) |>
   dplyr::filter(!is.na(.data$PFT))
 pft_by_code <- pft_lookup |>
@@ -61,17 +106,17 @@ crops <- DBI::dbGetQuery(con, sprintf(
           CAST(season AS INTEGER) AS season, CLASS,
           TRY_CAST(NULLIF(NULLIF(TRIM(CAST(SUBCLASS AS VARCHAR)), '**'), '') AS INTEGER) AS SUBCLASS
    FROM read_parquet('%s') WHERE \"year\" IN (%s) AND CLASS IS NOT NULL",
-  config[["crops_path"]], yr_list)) |>
+  crops_path, yr_list)) |>
   dplyr::rename(year = "yr") |>
   dplyr::mutate(code = paste0(.data$CLASS, .data$SUBCLASS))
 
 anchor_cols <- sort(unique(timing$anchor_col))
-phen_glob <- file.path(config[["phen_dir"]], config[["phen_glob"]])
+phen_files <- file.path(phen_dir, phen_glob)
 phen <- DBI::dbGetQuery(con, sprintf(
   "SELECT CAST(parcel_id AS INTEGER) AS parcel_id, CAST(\"year\" AS INTEGER) AS year,
           CAST(season AS INTEGER) AS season, gapfill_date_source, %s
    FROM read_parquet('%s') WHERE \"year\" IN (%s)",
-  paste(anchor_cols, collapse = ", "), phen_glob, yr_list))
+  paste(anchor_cols, collapse = ", "), phen_files, yr_list))
 
 missing_cols <- setdiff(anchor_cols, names(phen))
 if (length(missing_cols) > 0) {

workflows/ncc-statewide/02-sample-ncc-events.R

@@ -3,9 +3,20 @@
 config <- config::get(file = "workflows/ncc-statewide/config.yml",
                       config = Sys.getenv("NCC_PROJECT", "default"))
 
+# an override set to an empty string is treated as unset
+ccmmf_dir <- Sys.getenv("CCMMF_DIR")
+if (!nzchar(ccmmf_dir)) {
+  ccmmf_dir <- config[["ccmmf_dir"]]
+}
+output_dir <- Sys.getenv("CCMMF_NCC_OUT")
+if (!nzchar(output_dir)) {
+  output_dir <- file.path(ccmmf_dir, config[["output_dir"]])
+}
+output_dir <- path.expand(output_dir)
+
 set.seed(config[["seed"]])
 
-staging_dir <- file.path(config[["output_dir"]], "_staging")
+staging_dir <- file.path(output_dir, "_staging")
 design_file <- file.path(staging_dir, "_staging_01_design.rds")
 if (!file.exists(design_file)) {
   PEcAn.logger::logger.severe(