hurricane_normalization_app/app/queries.R

# DB queries for shiny app

library(tidyverse)
library(dplyr)
library(DBI)
library(xts)
library(pool)


config <- config::get(file = "config.yml")

# SUPABASE CON
con <- dbPool(
  RPostgres::Postgres(),
  host = config$db_host,
  port = config$db_port,
  dbname = config$db_dbname,
  user = config$db_user,
  password = config$db_password
)

#con <- dbConnect(
#  RPostgres::Postgres(),
#  host = config$db_host,
#  port = config$db_port,
#  dbname = config$db_dbname,
#  user = config$db_user,
#  password = config$db_password
#)

#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 <- tbl(con, I("econ.usa_yearly")) %>%
    filter(year == yr)

  result <- query %>% collect()

  return(result)
}

#get_latest_normalization_year <- function() {
#  query <- tbl(con, "all_yearly_normalized_losses") %>%
#    summarize(latest_year = max(normalization_year))
#
#  result <- query %>% collect()
#
#  return(result)
#}

get_storm_data_coverage <- function() {
  query <- tbl(con, "storm_data_coverage")

  result <- query %>% collect()

  return(result)
}

get_yearly_usa_pop_hu <- function(yr) {
  query <- tbl(con, I("metrics.usa_pop_hu")) %>%
    filter(year == yr)

  result <- query %>% collect()

  return(result)
}

get_best_track_summary <- function(storm) {
  query <- tbl(con, "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 <- tbl(con, "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 <- tbl(con, "all_loss_storms")

  result <- query %>% collect()

  return(result)
}

# returns a list of all stored hurdat ids
get_all_hurdat_ids <- function() {
  query <- tbl(con, "hurdat_ids")

  result <- query %>% collect()

  return(result)
}

# returns a list of latest normalized losses
get_latest_aggregate_losses <- function() {
  query <- tbl(con, "all_normalized_losses")

  result <- query %>% collect()

  return(result)
}

get_latest_aggregate_loss <- function(storm) {
  query <- tbl(con, "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 <- tbl(con, "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 <- tbl(con, "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)
}

# returns normalized mmh/mmp indexes and costs over time by storm
get_all_normalized_cost_index <- function(storm) {
  query <- tbl(con, "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 <- tbl(con, "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 <- tbl(con, "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 <- tbl(con, "all_loss_storms_tracks")

  result <- query %>% collect()

  return(result)
}

# returns storm track from HURDAT
get_hurdat_track <- function(storm) {
  query <- tbl(con, "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 <- tbl(con, I("gis.affected_area_landfalls")) %>%
      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 <- tbl(con, I("metrics.pop_housing_normalized_growth")) %>%
      #filter(base_year_population == storm$storm_year) %>%
      inner_join(affected_counties, by = c("state_fips", "county_fips")) %>%
      inner_join(
        tbl(con, I("public.counties")),
        by = c("state_fips" = "statefp", "county_fips" = "countyfp")
      ) %>%
      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 <- tbl(con, I("gis.indirect_landfalls")) %>%
      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 <- tbl(con, I("metrics.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 <- tbl(con, I("metrics.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(
        tbl(con, I("gis.us_states_boundary")) %>%
          rename(state_fips = statefp),
        by = "state_fips"
      ) %>%
      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 <- tbl(con, "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 <- tbl(con, "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 <- tbl(con, "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 <- tbl(con, I("public.counties")) %>%
    filter(
      statefp %in% !!unique_state_fips,
      countyfp %in% !!unique_county_fips
    ) %>%
    select(
      statefp,
      countyfp,
      name,
      namelsad
    ) %>%
    collect()

  states <- tbl(con, I("fips.states")) %>%
    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)
}

get_fatality_heatmap_data <- function(storm) {
  query <- tbl(con, I("fatal.conus_direct_fatalities")) %>%
    filter(
      storm_basin == storm$storm_basin,
      storm_year == storm$storm_year,
      storm_name == storm$storm_name
    ) %>%
    select(state_fips, fatality_type, n = fatality_count)

  result <- query %>%
    collect() %>%
    left_join(
      tbl(con, I("fips.states")) %>%
        select(state_fips, state_abbreviation) %>%
        collect(),
      by = "state_fips"
    ) %>%
    mutate(
      state_label = if_else(state_fips == "00", "Unknown", state_abbreviation)
    )

  return(result)
}

get_fatality_map <- function(storm) {
  query <- tbl(con, I("fatal.conus_direct_fatalities")) %>%
    filter(
      storm_basin == storm$storm_basin,
      storm_year == storm$storm_year,
      storm_name == storm$storm_name
    ) %>%
    inner_join(
      tbl(con, I("gis.us_states_boundary")) %>%
        rename(state_fips = statefp),
      by = "state_fips"
    ) %>%
    mutate(
      geom_wkt = sql(
        "ST_AsText(ST_SimplifyPreserveTopology(ST_Transform(wkb_geometry, 4326), .001))"
      )
    )

  result <- query %>% collect()

  return(result)
}