# update.packages(repos = "https://cran.rstudio.com/", # ask = FALSE) # # install.packages("pak", # repos = "https://mac.r-project.org") # # options("pkg.cran_mirror" = "https://mac.r-project.org") # # # installed.packages() |> # # rownames() |> # # pak::pkg_install(upgrade = TRUE, # # ask = FALSE) # # pak::pak( # c( # "arrow?source", # "sf?source", # "curl", # "tidyverse", # "tigris", # "rmapshaper", # "furrr", # "future.mirai" # ) # ) library(magrittr) library(tidyverse) library(sf) library(arrow) library(furrr) library(future.mirai) source("R/s3-archive.R") s3_preflight() s3_bucket_name <- Sys.getenv("S3_BUCKET", unset = "sustainable-fsa") s3_prefix <- Sys.getenv("S3_PREFIX", unset = "usdm-counties-census-2020") ## Pull prior archive state so incremental guards see existing outputs s3_pull(s3_bucket_name, paste0(s3_prefix, "/data"), "data") dir.create( file.path("data", "usdm"), recursive = TRUE, showWarnings = FALSE ) ## Load the 2020 TIGER/Line county boundaries from the census-counties archive, ## which owns the download, the schema normalization and the validity repair. ## ## That archive does not carry the state name — it publishes the Census schema ## — so join it here. The determinations below are written with a State column, ## and the boundaries used to arrive with one because usdm-counties joined it ## before republishing them. states <- tigris::states(cb = TRUE) %>% sf::st_drop_geometry() %>% dplyr::select(STATEFP, State = NAME) %>% dplyr::arrange(STATEFP) if(!file.exists("data/census-2020-counties.parquet")){ sf::read_sf( "https://data.sustainable-fsa.com/census-counties/data/parquet/2020-counties.parquet" ) %>% dplyr::filter(!(STATEFP %in% c("60", "66", "69", "78"))) %>% dplyr::left_join(states, by = dplyr::join_by(STATEFP)) %>% dplyr::select(STATEFP, State, COUNTYFP, County, `CountyLSAD`, Area) %>% sf::write_sf( "data/census-2020-counties.parquet", driver = "Parquet", layer_options = c("COMPRESSION=ZSTD", "COMPRESSION_LEVEL=13"), delete_dsn = TRUE ) } counties <- sf::read_sf("data/census-2020-counties.parquet") %>% sf::`st_agr<-`("constant") ## Get the current list of USDM dates usdm_get_dates <- function(as_of = lubridate::today("America/Denver")){ as_of %<>% lubridate::as_date() usdm_dates <- seq(lubridate::as_date("20000104"), lubridate::today(), "1 week") usdm_dates <- usdm_dates[(as_of - usdm_dates) >= 2] return(usdm_dates) } plan(mirai_multisession) usdm_get_dates() %>% tibble::tibble(Date = .) %>% dplyr::mutate( USDM = file.path("https://data.sustainable-fsa.com/usdm", "data", "parquet", paste0("USDM_",Date,".parquet")), outfile = file.path("data", "usdm", paste0("USDM_",Date,".parquet")) ) %>% dplyr::filter(!file.exists(outfile)) %>% ## Freshness gate: drop weeks whose upstream usdm parquet isn't published ## yet (fallback/premature runs no-op instead of failing in read_sf). (function(df){ posted <- purrr::map_lgl(df$USDM, url_exists) purrr::walk(df$Date[!posted], function(d) gate_skip(paste0("Upstream usdm parquet for ", d, " not yet published; skipping."))) df[posted, ] }) %>% furrr::future_pwalk( .f = function(USDM, outfile, ...){ if(!file.exists(outfile)){ usdm <- USDM %>% sf::read_sf() %>% sf::st_transform(sf::st_crs(counties)) %>% sf::`st_agr<-`("constant") dplyr::bind_rows( sf::st_intersection( counties, usdm ), sf::st_difference( counties, usdm %>% sf::st_union() ) ) %>% tidyr::fill(date) %>% sf::st_cast("MULTIPOLYGON") %>% sf::st_make_valid() %>% dplyr::arrange(STATEFP, COUNTYFP, date, usdm_class) %>% dplyr::mutate( usdm_date = date, usdm_class = tidyr::replace_na(usdm_class, "None") %>% factor(levels = c("None", paste0("D", 0:4)), ordered = TRUE), usdm_percent = units::drop_units(sf::st_area(geometry) / Area) ) %>% dplyr::select(STATEFP, State, COUNTYFP, County, CountyLSAD, usdm_date, usdm_class, usdm_percent) %>% dplyr::arrange(STATEFP, COUNTYFP, usdm_class) %>% sf::st_drop_geometry() %>% arrow::write_parquet(sink = outfile, version = "latest", compression = "zstd", compression_level = 13, use_dictionary = TRUE) } } ) plan(sequential) ## Create a single parquet output, for simplicity list.files("data/usdm", recursive = TRUE, full.names = TRUE) %>% purrr::map_dfr(arrow::read_parquet) %>% dplyr::arrange(STATEFP, COUNTYFP, usdm_date, usdm_class) %>% arrow::write_parquet(sink = "usdm-counties-census-2020.parquet", version = "latest", compression = "zstd", compression_level = 13, use_dictionary = TRUE) ## Create directory listing infrastructure generate_tree_flat <- function( data_dir = "data", output_file = file.path("manifest.json")) { all_entries <- fs::dir_ls(data_dir, recurse = TRUE, all = TRUE, type = "file") |> stringr::str_subset("(^|/)[.][^/]+", negate = TRUE) entries <- list() for (entry in all_entries) { rel_path <- fs::path_rel(entry, start = ".") info <- fs::file_info(entry) is_dir <- fs::is_dir(entry) entry_data <- list( path = as.character(rel_path), size = if (is_dir) "-" else info$size, mtime = if (is_dir) "-" else format(info$modification_time, "%Y-%Om-%d %H:%M:%S") ) entries[[length(entries) + 1]] <- entry_data } # Sort by path entries <- entries[order(sapply(entries, function(x) x$path))] jsonlite::write_json(entries, output_file, pretty = TRUE, auto_unbox = TRUE) message("✅ Wrote ", length(entries), " entries to ", output_file) } # Generate the flat index generate_tree_flat() ## ---- Publish to S3 --------------------------------------------------- s3_push(s3_bucket_name, paste0(s3_prefix, "/data"), "data", delete = TRUE) s3_put(s3_bucket_name, paste0(s3_prefix, "/usdm-counties-census-2020.parquet"), "usdm-counties-census-2020.parquet", content_type = "application/vnd.apache.parquet", cache_control = "max-age=3600") s3_put(s3_bucket_name, paste0(s3_prefix, "/manifest.json"), "manifest.json", content_type = "application/json", cache_control = "max-age=3600") s3_verify(s3_bucket_name, paste0(s3_prefix, "/data"), "data", allow_extra = character(0)) s3_write_manifest(s3_bucket_name, s3_prefix) cf_invalidate(c(paste0("/", s3_prefix, "/usdm-counties-census-2020.parquet"), paste0("/", s3_prefix, "/manifest.json"), paste0("/", s3_prefix, "/_manifest.txt"))) # ---- Render the README ---- # Regenerates README.md and the example map from the freshly updated # archive; the workflow commits these (and only these) back to git. cf_wait_manifest("https://data.sustainable-fsa.com/usdm-counties-census-2020/manifest.json", "manifest.json") rmarkdown::render("README.Rmd")