utils.py

import os
import folium
import confuse
import numpy as np
from math import isnan
import geopandas as gpd
from shapely.geometry import Point
from PIL import Image
from tqdm import tqdm

# Initialzie custom basemaps for folium
basemaps = {
    'Google Maps': folium.TileLayer(
        tiles = 'https://mt1.google.com/vt/lyrs=m&x={x}&y={y}&z={z}',
        attr = 'Google',
        name = 'Google Maps',
        overlay = True,
        control = True
    ),
    'Google Satellite': folium.TileLayer(
        tiles = 'https://mt1.google.com/vt/lyrs=s&x={x}&y={y}&z={z}',
        attr = 'Google',
        name = 'Google Satellite',
        overlay = True,
        control = True
    ),
    'Google Terrain': folium.TileLayer(
        tiles = 'https://mt1.google.com/vt/lyrs=p&x={x}&y={y}&z={z}',
        attr = 'Google',
        name = 'Google Terrain',
        overlay = True,
        control = True
    ),
    'Google Satellite Hybrid': folium.TileLayer(
        tiles = 'https://mt1.google.com/vt/lyrs=y&x={x}&y={y}&z={z}',
        attr = 'Google',
        name = 'Google Satellite',
        overlay = True,
        control = True
    ),
    'Esri Satellite': folium.TileLayer(
        tiles = 'https://server.arcgisonline.com/ArcGIS/rest/services/World_Imagery/MapServer/tile/{z}/{y}/{x}',
        attr = 'Esri',
        name = 'Esri Satellite',
        overlay = True,
        control = True
    ),
    'openstreetmap': folium.TileLayer('openstreetmap'),
    'cartodbdark_matter': folium.TileLayer('cartodbdark_matter')
}


# Dictionary of JavaScript files (More Readable)
scripts_dir = './scripts/'
scripts_files = [f for f in os.listdir(scripts_dir) if f.endswith('.js')]
Scripts = {}
for f in scripts_files:
    key = f.split('.')[0].upper()
    with open(scripts_dir + f) as f:
        Scripts[key] = f.read()

def calculate_bbox(df, field):
    '''
    Calculate the bounding box of a specfic field  ID in a  given data frame
    '''
    bbox = df.loc[df['name'] == field].bounds
    r = bbox.iloc[0]
    return [r.minx, r.miny, r.maxx, r.maxy]

def tiff_to_geodataframe(im, metric, date, crs):
    '''
    Convert a tiff image to a geodataframe
    '''
    x_cords = im.coords['x'].values
    y_cords = im.coords['y'].values
    vals = im.values
    dims = vals.shape
    points = []
    v_s = []
    for lat in range(dims[1]):
        y = y_cords[lat]
        for lon in range(dims[2]):
            x = x_cords[lon]
            v = vals[:,lat,lon]
            if isnan(v[0]):
                continue
            points.append(Point(x,y))
            v_s.append(v.item())   
    d = {f'{metric}_{date}': v_s, 'geometry': points} 
    df = gpd.GeoDataFrame(d, crs = crs)
    return df

def get_bearer_token_headers(bearer_token):
    '''
    Get the bearer token headers to be used in the request to the SentinelHub API
    '''
    headers = {
    'Content-Type': 'application/json',
    'Authorization': 'Bearer '+ bearer_token,
    }
    return headers

def get_downloaded_location_img_path(clientName, metric, date, field, extension='tiff'):
    '''
    Get the path of the downloaded image in TIFF based on the:
    '''
    date_dir = f'./{clientName}/raw/{metric}/{date}/field_{field}/'
    print(f'True Color Date Dir: {date_dir}')
    os.makedirs(date_dir, exist_ok=True)
    intermediate_dirs = os.listdir(date_dir)
    print(f'Intermediate Dirs: {intermediate_dirs}')
    if len(intermediate_dirs) == 0:
        return None
    imagePath = f'{date_dir}{os.listdir(date_dir)[0]}/response.{extension}'
    print(f'Image Path: {imagePath}')
    if not os.path.exists(imagePath):
        return None
    print(f'Image Path: {imagePath}')
    return imagePath

def get_masked_location_img_path(clientName, metric, date, field):
    '''
    Get the path of the downloaded image after applying the mask in TIFF based on the:
    '''
    date_dir = f'./{clientName}/processed/{metric}/{date}/field_{field}/'
    imagePath = date_dir + 'masked.tiff'
    return imagePath

def get_curated_location_img_path(clientName, metric, date, field):
    '''
    Get the path of the downloaded image after applying the mask and converting it to geojson formay based on the:
    '''
    date_dir = f'./{clientName}/curated/{metric}/{date}/field_{field}/'
    imagePath = date_dir + 'masked.geojson'

    if os.path.exists(imagePath):
        return imagePath
    else:
        return None

def parse_app_config(path=r'config-fgm-dev.yaml'):
    config = confuse.Configuration('CropHealth', __name__)
    config.set_file(path)
    return config


def fix_image(img):
    def normalize(band):
        band_min, band_max = (band.min(), band.max())
        return ((band-band_min)/((band_max - band_min)))
    def brighten(band):
        alpha=3
        beta=0
        return np.clip(alpha*band+beta, 0,255)
    def gammacorr(band):
        gamma=0.9
        return np.power(band, 1/gamma)
    red   = img[:, :, 0]
    green = img[:, :, 1]
    blue  = img[:, :, 2]
    red_b=brighten(red)
    blue_b=brighten(blue)
    green_b=brighten(green)
    red_bg=gammacorr(red_b)
    blue_bg=gammacorr(blue_b)
    green_bg=gammacorr(green_b)
    red_bgn = normalize(red_bg)
    green_bgn = normalize(green_bg)
    blue_bgn = normalize(blue_bg)
    rgb_composite_bgn= np.dstack((red_b, green_b, blue_b))
    return rgb_composite_bgn


def creat_gif(dataset, gif_name, duration=50):
    '''
    Create a gif from a list of images
    '''
    imgs = [Image.fromarray((255*img).astype(np.uint8)) for img in dataset]
    # duration is the number of milliseconds between frames; this is 40 frames per second
    imgs[0].save(gif_name, save_all=True, append_images=imgs[1:], duration=duration, loop=1)


def add_lat_lon_to_gdf_from_geometry(gdf):
    gdf['Lat'] = gdf['geometry'].apply(lambda p: p.x)
    gdf['Lon'] = gdf['geometry'].apply(lambda p: p.y)
    return gdf

def gdf_column_to_one_band_array(gdf, column_name):
    gdf = gdf.sort_values(by=['Lat', 'Lon'])
    gdf = gdf.reset_index(drop=True)
    unique_lats_count = gdf['Lat'].nunique()
    unique_lons_count = gdf['Lon'].nunique()
    rows_arr = [[] for i in range(unique_lats_count)]
    column_values = gdf[column_name].values
    for i in tqdm(range(len(column_values))):
        row_index = i // unique_lons_count
        rows_arr[row_index].append(column_values[i])

    max_row_length = max([len(row) for row in rows_arr])
    for row in rows_arr:
        while len(row) < max_row_length:
            row.append(0)

    rows_arr = np.array(rows_arr)
    return rows_arr