# DB queries for shiny app

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

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
)

# lazy load 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"))

# test storm
test_storm <- list(
  storm_basin = "AL",
  storm_year = 2005,
  storm_name = "KATRINA",
  lf_type = "LF",
  lf_id = 1
)

# helper functions

# 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 <- "SELECT * FROM all_loss_storms"
  
  result <- dbGetQuery(con, query)
  
  return(result)
}

# returns a list of latest normalized losses
get_latest_aggregate_losses <- function() {
  query <- "SELECT * FROM all_normalized_losses"
  
  result <- dbGetQuery(con, query)
  
  return(result)
}

# returns unique lf ids for a storm
get_unique_lf_ids <- function(storm) {
  query <- econ.storm_base_loss %>% 
    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)
    ) %>% 
    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 <- econ.normalized_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
    ) %>% 
    left_join(econ.storm_base_loss %>% 
                filter(!is.na(base_loss)) %>% 
                mutate(
                  ncei_priority = case_when(
                    str_like(base_loss_source, "%ncei%") ~ 1,
                    str_like(base_loss_source, "%mwr%") ~ 2,
                    TRUE ~ 3
                  )
                ) %>% 
                group_by(storm_basin, storm_year, storm_name, lf_type, lf_id) %>% 
                slice_min(ncei_priority, n = 1, with_ties = F) %>% 
                ungroup(),
              by = c("storm_basin", "storm_year", "storm_name", "lf_type", "lf_id")) %>% 
      mutate(
        mmh_index = (gdp_deflator * rwhu * affected_housing),
        mmp_index = (gdp_deflator * rwpc * affected_population),
        mmh = (base_loss * mmh_index),
        mmp = (base_loss * mmp_index)
      ) %>% 
      select(
        normalization_year, mmh, mmp
      )
  
  result <- query %>% collect()
  
  return(result)
}

# returns landfalls and data at landfall from HURDAT
get_hurdat_landfalls <- function(storm) {
  query <- hurdat.best_track %>% 
    filter(
      storm_basin == storm$storm_basin,
      storm_year == storm$storm_year,
      storm_name == storm$storm_name,
      record_identifier == "L"
    ) %>% 
    mutate(
      lon = sql("ST_X(ST_Transform(location::geometry, 4326))"),
      lat = sql("ST_Y(ST_Transform(location::geometry, 4326))")
    ) %>% 
    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 storm track from HURDAT
get_hurdat_track <- function(storm) {
  query <- hurdat.best_track %>% 
    filter(
      storm_basin == storm$storm_basin,
      storm_year == storm$storm_year,
      storm_name == storm$storm_name
    ) %>% 
    mutate(
      lon = sql("ST_X(ST_Transform(location::geometry, 4326))"),
      lat = sql("ST_Y(ST_Transform(location::geometry, 4326))"),
      rmw_meters = (rmw * 1852) # convert nautical miles to meters
    ) %>% 
    select(
      datetime,
      lon,
      lat,
      rmw,
      record_identifier,
      rmw_meters
    )
  
  result <- query %>% collect()
  
  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_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)
}

async_db_query <- function(query_func, ...) {
  args <- list(...)
  
  mirai_call <- mirai({
    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"))
    
    source(file = paste0(baseDir, "R/dataScripts/restructured/app/queries.R"))
    
    if(length(args) == 0) {
      query_func()
    }else{
      do.call(query_func, args)
    }
  },
  environment()
  )
  
  return(mirai_call)
}

# test functions

#storm <- list(storm_basin = "AL", storm_name = "KATRINA", storm_year = 2005)
#full_lf_id <- "LF1"

#a <- get_all_loss_storms()

#b <- get_latest_aggregate_losses()

#c <- get_unique_lf_ids(storm)

#d <- get_normalized_cost_index(storm, full_lf_id)

#e <- get_hurdat_landfalls(storm)

#f <- get_hurdat_track(storm)

#g <- get_normalized_metric_growth(storm, full_lf_id)
#g_2 <- g %>% 
#  st_as_sf(wkt = "geom_wkt")