using CSV
using NighttimeLights
using Rasters
using Plots
using DataFrames
using Dates
using MySQL
using DataFrames

function agg_radiance_annual(lat, long, R = 10, res = 15)
    raster1 = Raster(radiance_path * readdir(radiance_path)[1])
    bounds, m = annular_ring(raster1, R, lat, long, res) 
    radiances = [Raster(i, lazy = true)[bounds...] for i in radiance_path .* readdir(radiance_path)]
    series = RasterSeries(radiances, Ti(timestamps))
    radiance_datacube = Rasters.combine(series, Ti)       
    return [sum(skipmissing(Array(radiance_datacube[:,:,1,i] .* m))) for i in 1:length(readdir(radiance_path))]
end 

function CapexIndustryNTLEventStudy(industry,status,event,start_date,end_date,outer_radius,inner_radius)
    host = ENV["DBIP"]
    user = ENV["HHD_AND_FLD_USER"] 
    password = ENV["HHD_AND_FLD_PASSWORD"]
    conn = DBInterface.connect(MySQL.Connection, host, user, password; db = "fld")
    project_event = DataFrame(DBInterface.execute(conn,"select COMPANY,PROJECT_NAME,PROJECT_NUMBER,
                                                    COMPANY_CODE,EVENT,EVENT_DATE from
                                                    capex_all_project_events"))
    rename!(project_event, lowercase.(names(project_event)))   
    # Getting project status data
    project_status = DataFrame(DBInterface.execute(conn,"select COMPANY,PROJECT_NAME,PROJECT_NUMBER,
                                                    COMPANY_CODE,PROJECT_STATUS,INDUSTRY from
                                                    capex_all_project_detail"))
    rename!(project_status, lowercase.(names(project_status)))
    filtered_project_status = filter(row -> (row[:industry] == industry) && (row[:project_status] == status), project_status)
    filtered_project_event = filter(row -> row.event == event, project_event)
    filtered_project_event.project_id = string.(filtered_project_event.project_name, "_", filtered_project_event.project_number, "_", filtered_project_event.company_code, "_", filtered_project_event.company)
    filtered_project_status.project_id = string.(filtered_project_status.project_name, "_", filtered_project_status.project_number, "_", filtered_project_status.company_code, "_", filtered_project_status.company)
    select!(filtered_project_event, Not([:company, :company_code, :project_number, :project_name]))
    select!(filtered_project_status, Not([:company, :company_code, :project_number, :project_name]))
    joined_df = leftjoin(filtered_project_status, filtered_project_event, on=:project_id)
    # Curing the date (year-month-date)
    joined_df.event_date = Dates.Date.(joined_df.event_date, "dd u yyyy")
    # filtering between start and end date for where the event can lie
    start_date = Date(start_date)
    end_date = Date(end_date)   
    date_filtered_df = filter(row -> start_date <= row.event_date <= end_date, joined_df)
    # event dataframe
    select!(date_filtered_df, Not([:project_status, :industry, :event]))
    #Now extracting the lat longs of the projects
    project_location = DataFrame(DBInterface.execute(conn,"select * from capex_all_project_location"))
    rename!(project_location, lowercase.(names(project_location)))
    project_location.project_id = string.(project_location.project_name, "_", project_location.project_number, "_", project_location.company_code, "_", project_location.company)
    select!(project_location, Not([:company, :company_code, :project_number, :project_name,:location,:district,:district_code,:state]))
    rename!(project_location,:"longitude,latitude" => :lat_long)
    # Now only keeping point projects and projects whose lat_long are given 
    filtered_lat_long = filter(row -> !ismissing(row.lat_long) && 
        count(i -> i == ',', row.lat_long) == 1 &&  
        count(i -> i == ':', row.lat_long) == 0 &&
        length(split(row.lat_long, ',')) == 2, project_location)
    joined_lat_long = leftjoin(date_filtered_df,filtered_lat_long,on=:project_id)
    unique_lat_long = unique(joined_lat_long[:, [:project_id, :lat_long]])
    unique_lat_long = dropmissing(unique_lat_long)
    # Now extracting the radiance values 
    radiance_path = "/home/nighttimelights/VIIRS_ANNUAL/v21_median/"
    dates = range(start = Date(2012), step = Year(1), length = length(readdir(radiance_path))) |> collect
    timestamps = year.(dates)
    aggs = []
    for i in unique_lat_long.lat_long
        lat, long = tuple(parse.(Float64,split(i, ","))...)
        push!(aggs, agg_radiance_annual(lat, long,outer_radius,inner_radius))
    end
    agg_matrix =  cat(aggs..., dims = 2)'
    agg_df = DataFrame(Array(agg_matrix), :auto)
    rename!(agg_df, Symbol.(dates))
    agg_df.lat_long = unique_lat_long.lat_long
    capexnl = leftjoin(unique_lat_long, agg_df, on = :lat_long)
    return Dict("events" => date_filtered_df, "values" => capexnl)
end