# DB queries for shiny app library(tidyverse) library(dplyr) library(DBI) library(xts) library(memoise) library(cachem) library(digest) APP_DIR <- getwd() config <- config::get(file = "config.yml") # SUPABASE CON con <- dbConnect( RPostgres::Postgres(), host = config$db_host, port = config$db_port, dbname = config$db_dbname, user = config$db_user, password = config$db_password ) .tbl_cache <- new.env(parent = emptyenv()) get_tbl <- function(name, schema = NULL) { key <- if (is.null(schema)) name else paste0(schema, ".", name) if (!exists(key, .tbl_cache)) { if (is.null(schema)) { .tbl_cache[[key]] <- tbl(con, name) } else { .tbl_cache[[key]] <- tbl(con, I(paste0(schema, ".", name))) } } .tbl_cache[[key]] } #qry <- econ.storm_base_loss %>% # left_join(view.hurdat_track, by = c("storm_basin", "storm_year", "storm_name")) #show_query(qry) # test storm #katrina <- list( # storm_basin = "AL", # storm_year = 2005, # storm_name = "KATRINA", # lf_type = "LF", # lf_id = 1 #) #galveston <- list( # storm_basin = "AL", # storm_year = 1900, # storm_name = "GALVESTON", # lf_type = "LF", # lf_id = 1 #) agnes <- list( storm_basin = "AL", storm_year = 1972, storm_name = "AGNES" ) # helper functions # disconnect db connection disconnect_db <- function() { dbDisconnect(con) } # splits full_lf_id into lf_type and lf_id split_full_lf_id <- function(full_lf_id) { lf_type = gsub('[0-9]+', '', full_lf_id) lf_id = gsub('[^0-9]', '', full_lf_id) return(list(lf_type = lf_type, lf_id = lf_id)) } # DB getters get_yearly_economics <- function(yr) { query <- get_tbl("usa_yearly", "econ") %>% filter(year == yr) result <- query %>% collect() return(result) } get_latest_normalization_year <- function() { query <- get_tbl("all_yearly_normalized_losses") %>% summarize(latest_year = max(normalization_year)) result <- query %>% collect() return(result) } get_storm_data_coverage <- function() { query <- get_tbl("storm_data_coverage") result <- query %>% collect() return(result) } get_yearly_usa_pop_hu <- function(yr) { query <- get_tbl("usa_pop_hu", "metrics") %>% filter(year == yr) result <- query %>% collect() return(result) } get_best_track_summary <- function(storm) { query <- get_tbl("best_track_summary") %>% filter( storm_basin == storm$storm_basin, storm_year == storm$storm_year, storm_name == storm$storm_name ) result <- query %>% collect() return(result) } get_hurdat_id <- function(storm) { query <- get_tbl("hurdat_ids") %>% filter( storm_basin == storm$storm_basin, storm_year == storm$storm_year, storm_name == storm$storm_name ) result <- query %>% collect() return(result) } # returns a list of loss storms we have data on get_all_loss_storms <- function() { query <- get_tbl("all_loss_storms") result <- query %>% collect() return(result) } # returns a list of all stored hurdat ids get_all_hurdat_ids <- function() { query <- get_tbl("hurdat_ids") result <- query %>% collect() return(result) } # returns a list of latest normalized losses get_latest_aggregate_losses <- function() { query <- get_tbl("all_normalized_losses") result <- query %>% collect() return(result) } get_latest_aggregate_loss <- function(storm) { query <- get_tbl("all_normalized_losses") %>% filter( storm_year == storm$storm_year, storm_name == storm$storm_name ) result <- query %>% collect() return(result) } # returns unique lf ids for a storm get_unique_lf_ids <- function(storm) { query <- get_tbl("all_loss_landfalls") %>% filter( storm_basin == storm$storm_basin, storm_year == storm$storm_year, storm_name == storm$storm_name ) %>% distinct( full_lf_id, .keep_all = T ) %>% arrange( full_lf_id ) %>% select( storm_basin, storm_year, storm_name, lf_type, lf_id, full_lf_id ) result <- query %>% collect() return(result) } # returns normalized mmh/mmp indexes and costs over time by landfall get_normalized_cost_index <- function(storm, full_lf_id) { lf_id_parts <- split_full_lf_id(full_lf_id) query <- get_tbl("all_yearly_normalized_losses") %>% filter( storm_basin == storm$storm_basin, storm_year == storm$storm_year, storm_name == storm$storm_name, lf_type == lf_id_parts$lf_type, lf_id == lf_id_parts$lf_id ) %>% select( normalization_year, mmh, mmp ) result <- query %>% collect() return(result) } test_query <- function() { result <- dbGetQuery( con, "WITH yearly_totals AS ( SELECT year, SUM(population) as total_population, SUM(housing_units) as total_housing_units FROM metrics.pop_and_housing WHERE state_fips = '12' AND county_fips IN ('011', '021', '086', '087') AND year BETWEEN 1926 AND 2024 GROUP BY year ), base_year AS ( SELECT total_population as base_population, total_housing_units as base_housing FROM yearly_totals WHERE year = (SELECT MIN(year) FROM yearly_totals) -- Uses first year in dataset as base ) SELECT year, total_population, total_housing_units, ROUND(total_population / base_population, 4) as population_index, ROUND(total_housing_units / base_housing, 4) as housing_index FROM yearly_totals CROSS JOIN base_year ORDER BY year;" ) result <- result %>% mutate( year = as.Date(paste0(year, "-01-01")) ) %>% select( year, population_index, housing_index ) result_ts <- result %>% select(-year) %>% xts(order.by = result$year) return(result_ts) } # returns normalized mmh/mmp indexes and costs over time by storm get_all_normalized_cost_index <- function(storm) { query <- get_tbl("all_yearly_normalized_losses") %>% filter( storm_basin == storm$storm_basin, storm_year == storm$storm_year, storm_name == storm$storm_name ) %>% mutate( full_lf_id = paste0(lf_type, lf_id) ) %>% select( normalization_year, mmh_index, mmp_index, mmh_loss = mmh, mmp_loss = mmp, full_lf_id ) result <- query %>% collect() return(result) } # returns landfalls and data at landfall from HURDAT get_hurdat_landfalls <- function(storm) { query <- get_tbl("hurdat_track") %>% filter( storm_basin == storm$storm_basin, storm_year == storm$storm_year, storm_name == storm$storm_name, record_identifier == "L" ) %>% select( datetime, lon, lat, rmw, pressure, windspeed ) result <- query %>% collect() #return( # list( # data = result, # storm_info = storm, # row_count = nrow(result), # timestamp = Sys.time() # ) #) return(result) } # returns all tracked storm conus landfalls get_all_conus_landfalls <- function() { query <- get_tbl("all_conus_landfalls") %>% select( storm_year, storm_name, lon, lat, rmw_meters ) result <- query %>% collect() return(result) } # returns all storm tracks from HURDAT get_all_hurdat_tracks <- function() { query <- get_tbl("all_loss_storms_tracks") result <- query %>% collect() return(result) } # returns storm track from HURDAT get_hurdat_track <- function(storm) { query <- get_tbl("hurdat_track") %>% filter( storm_basin == storm$storm_basin, storm_year == storm$storm_year, storm_name == storm$storm_name ) %>% select( datetime, storm_status, lon, lat, rmw, pressure, windspeed, record_identifier, rmw_meters ) result <- query %>% collect() %>% mutate( formatted_datetime = paste0( format(datetime, "%m/%d/%Y"), " ", format(datetime, "%H"), "Z" ) ) %>% select( formatted_datetime, datetime, storm_status, lon, lat, rmw, pressure, windspeed, record_identifier, rmw_meters ) return(result) } # returns normalized population and housing growth by county with geometry get_normalized_metric_growth <- function(storm, full_lf_id) { lf_id_parts <- split_full_lf_id(full_lf_id) if (lf_id_parts$lf_type == 'LF') { affected_counties <- get_tbl("affected_area_landfalls", "gis") %>% filter( storm_basin == storm$storm_basin, storm_year == storm$storm_year, storm_name == storm$storm_name, lf_type == lf_id_parts$lf_type, lf_id == lf_id_parts$lf_id ) %>% select(state_fips, county_fips) query <- get_tbl("pop_housing_normalized_growth", "metrics") %>% filter(base_year == storm$storm_year) %>% inner_join(affected_counties, by = c("state_fips", "county_fips")) %>% inner_join( get_tbl("counties", "public"), by = c("state_fips" = "statefp", "county_fips" = "countyfp") ) %>% mutate( geom_wkt = sql( "ST_AsText(ST_SimplifyPreserveTopology(ST_Transform(geom, 4326), .001))" ) ) %>% select( state_fips, county_fips, year, name, population, housing_units, base_year_population, base_year_housing, normalized_population = population_normalized, normalized_housing = housing_units_normalized, geom_wkt ) %>% arrange(state_fips, county_fips, year) result <- query %>% collect() %>% mutate( normalized_population = if_else( is.na(base_year_population) | is.na(base_year_housing), NA_real_, normalized_population / 100 ), normalized_housing = if_else( is.na(base_year_population) | is.na(base_year_housing), NA_real_, normalized_housing / 100 ) ) %>% select(-base_year_population, -base_year_housing) return(result) } else { affected_state <- get_tbl("indirect_landfalls", "gis") %>% filter( storm_basin == storm$storm_basin, storm_year == storm$storm_year, storm_name == storm$storm_name, lf_type == lf_id_parts$lf_type, lf_id == lf_id_parts$lf_id ) %>% select( state_fips ) baseline_metrics <- get_tbl("yearly_state_metrics") %>% filter(year == storm$storm_year) %>% inner_join(affected_state, by = "state_fips") %>% select( state_fips, baseline_population = state_population, baseline_housing = state_housing_units ) normalized_metrics <- get_tbl("yearly_state_metrics") %>% filter( year >= storm$storm_year ) %>% inner_join(affected_state, by = "state_fips") %>% inner_join(baseline_metrics, by = "state_fips") %>% mutate( normalized_population = as.numeric(state_population) / as.numeric(baseline_population), normalized_housing = as.numeric(state_housing_units) / as.numeric(baseline_housing) ) %>% select( state_fips, year, population = state_population, housing_units = state_housing_units, normalized_population, normalized_housing ) query <- normalized_metrics %>% inner_join( get_tbl("us_states_boundary", "gis") %>% rename(state_fips = statefp), by = "state_fips" ) %>% mutate( geom_wkt = sql( "ST_AsText(ST_SimplifyPreserveTopology(ST_Transform(wkb_geometry, 4326), .001))" ) ) %>% select( state_fips, year, name, population, housing_units, normalized_population, normalized_housing, geom_wkt ) %>% arrange(state_fips, year) } result <- query %>% collect() return(result) } # returns all lf type landfalls get_all_lf_type_landfalls <- function() { query <- get_tbl("lf_id_location") result <- query %>% collect() return(result) } # returns all storms and factors needed to normalize a lf type landfall get_all_lf_type_factors <- function() { query <- get_tbl("lf_landfall_gis") result <- query %>% collect() return(result) } # returns one storm and factors needed to normalize a lf type landfall get_lf_type_factors <- function(storm) { query <- get_tbl("lf_landfall_gis") %>% filter( storm_basin == storm$storm_basin, storm_year == storm$storm_year, storm_name == storm$storm_name, full_lf_id == storm$full_lf_id ) result <- query %>% collect() return(result) } get_county_and_state <- function(df) { unique_state_fips <- unique(df$state_fips) unique_county_fips <- unique(df$county_fips) counties <- get_tbl("counties", "public") %>% filter( statefp %in% !!unique_state_fips, countyfp %in% !!unique_county_fips ) %>% select( statefp, countyfp, name, namelsad ) %>% collect() states <- get_tbl("states", "fips") %>% filter(state_fips %in% !!unique_state_fips) %>% select( state_fips, state_name, state_abbreviation ) %>% collect() result <- df %>% left_join( counties, by = c("state_fips" = "statefp", "county_fips" = "countyfp") ) %>% left_join( states, by = "state_fips" ) return(result) } cache_dir <- file.path(APP_DIR, "cache") cache_logfile <- file.path(APP_DIR, "cachelog") if (!dir.exists(cache_dir)) { dir.create(cache_dir, recursive = TRUE) } query_cache <- cachem::cache_disk( dir = cache_dir, max_age = Inf, evict = "lru", logfile = cache_logfile ) get_latest_normalization_year <- memoise( get_latest_normalization_year, cache = query_cache ) get_all_loss_storms <- memoise(get_all_loss_storms, cache = query_cache) get_all_hurdat_ids <- memoise(get_all_hurdat_ids, cache = query_cache) get_latest_aggregate_losses <- memoise( get_latest_aggregate_losses, cache = query_cache ) get_all_conus_landfalls <- memoise(get_all_conus_landfalls, cache = query_cache) get_all_hurdat_tracks <- memoise(get_all_hurdat_tracks, cache = query_cache) get_all_lf_type_factors <- memoise(get_all_lf_type_factors, cache = query_cache)