Merge pull request #54 from ccmmf/events-update

Event building: filtered and canonical

Author: infotroph

examples/3_rowcrop/example_user_config.yaml

@@ -35,3 +35,12 @@ aws_profile: magic
 # Stage example-specific input files into run_dir before prepare runs.
 external_paths:
   template_file: "examples/3_rowcrop/template.xml"
+
+paths:
+  field_shape_path: "data_raw/management/crops/v4.1/parcels-consolidated.gpkg"
+  phenology_info_path: "data_raw/management/phenology/v1.0/"
+  planting_info_path: "data_raw/management/planting/v1.0/"
+  harvest_info_path: "data_raw/management/harvest/v1.0/"
+  tillage_info_path: "data_raw/management/tillage/v1.0/"
+  irrigation_info_path: "data_raw/management/irrigation/v1.1/"
+

tools/event_prep/01a-clean-irrigation.R

@@ -17,6 +17,10 @@ options <- list(
       "Format will be Parquet organized by ensemble member,",
       "with subdirectories named in 'hive partion' style."
     )
+  ),
+  optparse::make_option("--site_info_path",
+    default = "site_info.csv",
+    help = "CSV giving ids to be extracted. Only column 'field_id' is used"
   )
 ) |>
   # Show default values in help message
@@ -31,6 +35,10 @@ args <- optparse::OptionParser(option_list = options) |>
 ## -------------------------- end option parsing ------------------------------
 
 dir.create(args$outdir, showWarnings = FALSE, recursive = TRUE)
+siteids <- read.csv(args$site_info_path) |>
+  _$field_id |>
+  unique() |>
+  glue::glue_collapse(",")
 
 dbdir <- tempfile("duckdb", fileext = ".duckdb")
 conn <- DBI::dbConnect(duckdb::duckdb(dbdir = dbdir))
@@ -58,6 +66,7 @@ DBI::dbExecute(conn, glue::glue("
       CAST (amount_mm AS DECIMAL(6, 2)) AS amount_mm,
       method
     FROM read_parquet('{args$irr_path}')
+    WHERE site_id IN ({siteids})
     ORDER BY event_member_id, site_id, date
   ) TO
   '{args$outdir}/irrigation.parquet' 

tools/event_prep/01b-clean-other-events.R

@@ -6,6 +6,10 @@
 
 ## ---------------------- parse command-line options --------------------------
 options <- list(
+  optparse::make_option("--site_info_path",
+    default = "site_info.csv",
+    help = "CSV giving ids to be extracted. Only column 'field_id' is used"
+  ),
   optparse::make_option("--pheno_dir",
     default = "data_raw/management/phenology/v1.0",
     help = "Directory containing Parquet files of phenology (leafon/leafoff) events"
@@ -53,6 +57,9 @@ args <- optparse::OptionParser(option_list = options) |>
 
 ## -------------------------- end option parsing ------------------------------
 
+siteids <- read.csv(args$site_info_path) |>
+  _$field_id |>
+  unique()
 
 harvest_files <- list.files(args$harvest_dir, "\\.parquet$", full.names = TRUE, recursive = TRUE)
 planting_files <- list.files(args$planting_dir, "\\.parquet$", full.names = TRUE, recursive = TRUE)
@@ -62,6 +69,7 @@ dir.create(args$outdir, showWarnings = FALSE, recursive = TRUE)
 
 message("Writing harvest output")
 harvest <- arrow::open_dataset(harvest_files, format = "parquet") |>
+dplyr::filter(as.character(site_id) %in% siteids) |>
   dplyr::mutate(
     site_id = as.integer(site_id),
     date = as.Date(date)
@@ -74,6 +82,7 @@ harvest <- arrow::open_dataset(harvest_files, format = "parquet") |>
 
 message("Writing planting output")
 planting <- arrow::open_dataset(planting_files, format = "parquet") |>
+  dplyr::filter(as.character(site_id) %in% siteids) |>
   dplyr::mutate(
     site_id = as.integer(site_id),
     date = pmax(as.Date(date), as.Date(args$adjust_start)) # push earlier plantings forward to avoid beginning-of-run boundary error
@@ -96,6 +105,7 @@ planting <- arrow::open_dataset(planting_files, format = "parquet") |>
 
 message("Writing tillage output")
 tillage <- arrow::open_dataset(tillage_files, format = "parquet") |>
+  dplyr::filter(as.character(site_id) %in% siteids) |>
   dplyr::filter(
     is.finite(.data$ndti_pct_change),
     .data$ndti_pct_change >= 0
@@ -121,6 +131,7 @@ tillage <- arrow::open_dataset(tillage_files, format = "parquet") |>
 message("Writing phenology output")
 phenology <- arrow::open_dataset(phenology_files, format = "parquet")
 leafon <- phenology |>
+  dplyr::filter(as.character(site_id) %in% siteids) |>
   dplyr::select("site_id", date = "leafonday") |>
   dplyr::mutate(
     site_id = as.integer(.data$site_id),
@@ -131,6 +142,7 @@ leafon <- phenology |>
     compression = "ZSTD"
   )
 leafoff <- phenology |>
+  dplyr::filter(site_id %in% siteids) |>
   dplyr::select("site_id", date = "leafoffday") |>
   dplyr::mutate(
     site_id = as.integer(.data$site_id),

examples/3_rowcrop/02a_build_events.R workflow/02a_build_events.Rexamples/3_rowcrop/02a_build_events.R renamed to workflow/02a_build_events.R

@@ -11,16 +11,27 @@ options <- list(
     default = "site_info.csv",
     help = "CSV giving ids, locations, and PFTs for sites of interest"
   ),
-  optparse::make_option("--raw_parquet_dir",
-    default = "data_raw/management",
-    help = paste(
-      "Directory containing management inputs in Parquet format.",
-      "Note: Code currently looks for subpaths that include hard-coded",
-      "version numbers for each management type.",
-      "If this path is empty, cleaning is skipped and clean Parquet files",
-      " must already exist at the path specified by --clean_parquet_dir`."
-    )
+  optparse::make_option("--phenology_path",
+    default = "data_raw/management/phenology/v1.0",
+    help = "Parquet file or directory of Parquet files containing phenology events",
+  ),
+  optparse::make_option("--planting_path",
+    default = "data_raw/management/planting/v1.0",
+    help = "Parquet file or directory of Parquet files containing planting events",
+  ),
+  optparse::make_option("--harvest_path",
+    default = "data_raw/management/harvest/v1.0",
+    help = "Parquet file or directory of Parquet files containing harvest events",
+  ),
+  optparse::make_option("--tillage_path",
+    default = "data_raw/management/tillage/v1.0",
+    help = "Parquet file or directory of Parquet files containing tillage events",
   ),
+  optparse::make_option("--irrigation_path",
+    default = "data_raw/management/irrigation/v1.1",
+    help = "Parquet file or directory of Parquet files containing tillage events",
+  ),
+  # TODO add fertilization and NCC here
   optparse::make_option("--clean_parquet_dir",
     default = "data/management_ensembles",
     help = paste(
@@ -60,19 +71,7 @@ this_file <- function() {
   cmd_args <- commandArgs(trailingOnly = FALSE)
   normalizePath(sub("^--file=", "", grep("^--file=", cmd_args, value = TRUE)))
 }
-event_prep_dir <- normalizePath(file.path(dirname(this_file()), "..", "..", "tools", "event_prep"))
-
-# TODO these probably deserve to be runtime args,
-# but first better fix other version-specific assumptions below
-mgmt_subdirs <- list(
-  pheno = file.path(args$raw_parquet_dir, "phenology/v1.0"),
-  plant = file.path(args$raw_parquet_dir, "planting/v1.0"),
-  harv = file.path(args$raw_parquet_dir, "harvest/v1.0"),
-  till = file.path(args$raw_parquet_dir, "tillage/v1.0"),
-  irri = file.path(args$raw_parquet_dir, "irrigation/v1.1"),
-  fert = NULL, # TODO
-  occ = NULL # TODO
-)
+event_prep_dir <- normalizePath(file.path(dirname(this_file()), "..", "tools", "event_prep"))
 
 if (!dir.exists(args$event_outdir)) {
   dir.create(args$event_outdir, recursive = TRUE)
@@ -82,28 +81,30 @@ cargs <- function(...) {
   paste0("--", ...names(), "=", list(...))
 }
 
-if (args$raw_parquet_dir != "") {
-  PEcAn.logger::logger.info("Cleaning irrigation files")
-  callr::rscript(
-    file.path(event_prep_dir, "01a-clean-irrigation.R"),
-    cmdargs = cargs(
-      irr_path = mgmt_subdirs$irri,
-      outdir = args$clean_parquet_dir
-    )
+
+PEcAn.logger::logger.info("Cleaning irrigation files")
+callr::rscript(
+  file.path(event_prep_dir, "01a-clean-irrigation.R"),
+  cmdargs = cargs(
+    irr_path = args$irrigation_path,
+    outdir = args$clean_parquet_dir,
+    site_info_path = args$site_info_path
   )
-  PEcAn.logger::logger.info("Cleaning other management files")
-  callr::rscript(
-    file.path(event_prep_dir, "01b-clean-other-events.R"),
-    cmdargs = cargs(
-      pheno_dir = mgmt_subdirs$pheno,
-      planting_dir = mgmt_subdirs$plant,
-      harvest_dir = mgmt_subdirs$harv,
-      tillage_dir = mgmt_subdirs$till,
-      adjust_start = args$start_date,
-      outdir = args$clean_parquet_dir
-    )
+)
+PEcAn.logger::logger.info("Cleaning other management files")
+callr::rscript(
+  file.path(event_prep_dir, "01b-clean-other-events.R"),
+  cmdargs = cargs(
+    site_info_path = args$site_info_path,
+    pheno_dir = args$phenology_path,
+    planting_dir = args$planting_path,
+    harvest_dir = args$harvest_path,
+    tillage_dir = args$tillage_path,
+    adjust_start = args$start_date,
+    outdir = args$clean_parquet_dir
   )
-}
+)
+
 
 PEcAn.logger::logger.info("converting management files to events")
 callr::rscript(

workflow/workflow_manifest.yaml

@@ -61,6 +61,24 @@ steps:
           outputs:
               site_sipnet_met_path: site_sipnet_met_path
 
+        - name: build-events
+          script: "workflow/02a_build_events.R"
+          r_libraries: [tidyverse, callr, glue]
+          inputs:
+              site_info_path: site_info_file
+              phenology_path: phenology_info_path
+              planting_path: planting_info_path
+              harvest_path: harvest_info_path
+              tillage_path: tillage_info_path
+              irrigation_path: irrigation_info_path
+          params:
+              start_date: start_date
+              end_date: end_date
+          outputs:
+              clean_parquet_dir: clean_parquet_dir
+              event_outdir: event_dir
+
+
         - name: build-ic
           script: "workflow/02_ic_build.R"
           r_libraries: [tidyverse]
@@ -253,11 +271,15 @@ steps:
               data_dir: data_dir
 
         - name: build-events
-          script: "examples/3_rowcrop/02a_build_events.R"
-          r_libraries: [tidyverse, callr]
+          script: "workflow/02a_build_events.R"
+          r_libraries: [tidyverse, callr, glue]
           inputs:
               site_info_path: site_info_file
-              raw_parquet_dir: raw_parquet_dir
+              phenology_path: phenology_info_path
+              planting_path: planting_info_path
+              harvest_path: harvest_info_path
+              tillage_path: tillage_info_path
+              irrigation_path: irrigation_info_path
           params:
               start_date: start_date
               end_date: end_date
@@ -309,7 +331,6 @@ paths:
     site_info_file: "site_info.csv"
     site_sipnet_met_path: "data/ERA5_SIPNET"
     site_era5_path: "data_raw/ERA5_CA_nc"
-    field_shape_path: "data_raw/management/crops/v4.1/parcels-consolidated.gpkg"
     data_dir: "data/IC_prep"
     ic_outdir: "IC_files"
     pft_dir: "data_raw/pfts"
@@ -322,7 +343,6 @@ paths:
     ic_dir: "IC_files"
     settings_xml: "settings.xml"
     config_xml: "output/pecan.CONFIGS.xml"
-    raw_parquet_dir: "data_raw/management"
     clean_parquet_dir: "data/management_ensembles"
     event_dir: "data/events"