hurricane_normalization_app/queries.R

# DB queries for shiny app

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

linuxdir <- "/home/dylan/Personal/Projects/Hurricane Normalization/"
macdir <- "~/Desktop/Personal/Projects/Hurricane Normalization/"
#baseDir <- macdir
baseDir <- linuxdir

config <- config::get(
  file = paste0(baseDir, "R/dataScripts/restructured/app/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
)

# DB TABLES

econ.normalized_landfalls <- tbl(con, I("econ.normalized_landfalls"))
econ.storm_base_loss <- tbl(con, I("econ.storm_base_loss"))
econ.usa_yearly <- tbl(con, I("econ.usa_yearly"))

fatal.storm_fatalities_type <- tbl(con, I("fatal.storm_fatalities_type"))
fatal.storm_total_fatalities <- tbl(con, I("fatal.storm_total_fatalities"))

fips.counties <- tbl(con, I("fips.counties"))
fips.states <- tbl(con, I("fips.states"))

gis.affected_area_landfalls <- tbl(con, I("gis.affected_area_landfalls"))

hurdat.best_track <- tbl(con, I("hurdat.best_track"))
hurdat.hurdat_storms <- tbl(con, I("hurdat.hurdat_storms"))

metrics.geo_attributes <- tbl(con, I("metrics.geo_attributes"))
metrics.pop_and_housing <- tbl(con, I("metrics.pop_and_housing"))

public.counties <- tbl(con, I("public.counties"))

# DB VIEWS
view.all_normalized_losses <- tbl(con, "all_normalized_losses")
view.hurdat_track <- tbl(con, "hurdat_track")
view.all_loss_storms <- tbl(con, "all_loss_storms")
view.all_loss_landfalls <- tbl(con, "all_loss_landfalls")
view.yearly_normalized_losses <- tbl(con, "yearly_normalized_losses")
view.simplified_county_geom <- tbl(con, "simplified_county_geom")
view.all_loss_storms_tracks <- tbl(con, "all_loss_storms_tracks")
view.all_conus_landfalls <- tbl(con, "all_conus_landfalls")

#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

# returns a list of loss storms we have data on
get_all_loss_storms <- function() {
  query <- view.all_loss_storms

  result <- query %>% collect()

  return(result)
}

# returns a list of latest normalized losses
get_latest_aggregate_losses <- function() {
  query <- view.all_normalized_losses

  result <- query %>% collect()

  return(result)
}

# returns unique lf ids for a storm
get_unique_lf_ids <- function(storm) {
  query <- view.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 <- view.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 <- view.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 <- view.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 <- view.all_conus_landfalls

  result <- query %>% collect()

  return(result)
}

# returns all storm tracks from HURDAT
get_all_hurdat_tracks <- function() {
  query <- view.all_loss_storms_tracks

  result <- query %>% collect()

  return(result)
}

# returns storm track from HURDAT
get_hurdat_track <- function(storm) {
  query <- view.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)

  affected_counties <- 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)

  baseline_metrics <- metrics.pop_and_housing %>%
    filter(
      year == storm$storm_year
    ) %>%
    inner_join(affected_counties, by = c("state_fips", "county_fips")) %>%
    select(
      state_fips,
      county_fips,
      baseline_population = population,
      baseline_housing = housing_units
    )

  normalized_metrics <- metrics.pop_and_housing %>%
    filter(
      year >= storm$storm_year
    ) %>%
    inner_join(affected_counties, by = c("state_fips", "county_fips")) %>%
    inner_join(baseline_metrics, by = c("state_fips", "county_fips")) %>%
    mutate(
      normalized_population = as.numeric(population) /
        as.numeric(baseline_population),
      normalized_housing = as.numeric(housing_units) /
        as.numeric(baseline_housing)
    ) %>%
    select(
      state_fips,
      county_fips,
      year,
      population,
      housing_units,
      normalized_population,
      normalized_housing
    )

  query <- normalized_metrics %>%
    inner_join(
      public.counties,
      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,
      population,
      housing_units,
      normalized_population,
      normalized_housing,
      geom_wkt
    ) %>%
    arrange(state_fips, county_fips, year)

  #query <- normalized_metrics %>%
  #  inner_join(
  #    public.counties,
  #    by = c("state_fips" = "statefp", "county_fips" = "countyfp")
  #  ) %>%
  #  mutate(
  #    geom_wkt = sql("ST_AsText(ST_Transform(geom, 4326))")
  #  ) %>%
  #  select(
  #    state_fips,
  #    county_fips,
  #    year,
  #    population,
  #    housing_units,
  #    normalized_population,
  #    normalized_housing,
  #    geom_wkt
  #  ) %>%
  #  arrange(state_fips, county_fips, year)

  result <- query %>% collect()

  return(result)
}