biomap/utils_gee.py

import io
import requests
import ee
import numpy as np
import matplotlib.pyplot as plt
import os
from pathlib import Path
import logging
import json

#Initialize
service_account = os.environ["SERVICE_ACCOUNT_EE"]
private_key = json.loads(os.environ["PRIVATE_KEY_EE"])

with open(os.path.join(os.path.dirname(__file__), '.private-key-2.json'), "w") as ipt:
    json.dump(private_key, ipt)

credentials = ee.ServiceAccountCredentials(service_account, os.path.join(os.path.dirname(__file__), '.private-key-2.json'))
ee.Initialize(credentials)

def get_image(location, d1, d2):
    logging.info(f"getting image for {d1} to {d2} at location {location}")
    img = extract_img(location, d1, d2)
    
    img_test = transform_ee_img(
        img, max=0.3
    )
    return img_test

#delete clouds
def maskS2clouds(image):
        qa = image.select('QA60');

    #   // Bits 10 and 11 are clouds and cirrus, respectively.
        cloudBitMask = 1 << 10;
        cirrusBitMask = 1 << 11;

    #   // Both flags should be set to zero, indicating clear conditions.
        mask = (qa.bitwiseAnd(cloudBitMask).eq(0))and(qa.bitwiseAnd(cirrusBitMask).eq(0))

        return image.updateMask(mask).divide(10000);


#find ee_img
def extract_ee_img(location,start_date,end_date, width = 0.01 , len = 0.01) :
    """Extract the earth engine image

    Args:
        location (list[float]): 
        start_date (str): the start date for finding an image
        end_date (str): the end date for finding an image
        width (float, optional): _description_. Defaults to 0.01.
        len (float, optional): _description_. Defaults to 0.01.

    Returns:
        _type_: _description_
    """
    # define the polygone
    polygone =[[[float(location[0])-0.01,float(location[1])+0.01],
                [float(location[0])-0.01,float(location[1])-0.01],
                [float(location[0])+0.01,float(location[1])-0.01],
                [float(location[0])+0.01,float(location[1])+0.01],
    ]]    

    #define the ee geometry 
    geometry = ee.Geometry.Polygon(polygone, None, False);

    #extract the dataset
    dataset = ee.ImageCollection('COPERNICUS/S2_SR_HARMONIZED')\
        .filterDate(start_date, end_date)\
        .filter(ee.Filter.lt('CLOUDY_PIXEL_PERCENTAGE',1))\
        .map(maskS2clouds)
    return dataset.mean(), geometry



# Get URL
def get_url(ee_img, geometry, scale=5):
    """Get the url of a dataset and a geometry 

    Args:
        ee_img (ee.ImageCollection: meta data on the image 
        geometry (ee.Geometry.Polygon): geometry of the desired landscape
        scale (int, optional): _description_. Defaults to 5.

    Returns:
        str: the url to use to ask the server
    """
    region = geometry

    # collectionList = ee_img.toList(ee_img.size())
    # collectionSize = collectionList.size().getInfo()
    # for i in xrange(collectionSize):
    #     ee.batch.Export.image.toDrive(
    #         image = ee.Image(collectionList.get(i)).clip(rectangle), 
    #         fileNamePrefix = 'foo' + str(i + 1), 
    #         dimensions = '128x128').start()

    url = ee_img.getDownloadURL({
        # 'min': 0.0,
        # 'max': 0.3,
        'bands': ['B4', 'B3', 'B2'],
        'region' : region,
        'scale' : scale,
        'format' : 'NPY'
    })
    
    return url

def extract_np_from_url(url):
    """extract a numpy array based on a url

    Args:
        url (str): _description_

    Returns:
        numpyarray: response from earth engine as numpy
    """
    #get the response from url
    response = requests.get(url)

    #transform it into numpy
    data = np.load(io.BytesIO(response.content))
    
    #transform numpy of tuples to 3D numpy
    temp1 = []

    for x in data:
        temp2 = []
        for y in x :
            temp2.append([z for z in y])
        temp1.append(temp2)

    data = np.array(temp1)
    return data
    
#Fonction globale
def extract_img(location,start_date,end_date, width = 0.01 , len = 0.01,scale=5):
    """Extract an image of the landscape at the selected longitude and latitude with the selected width and length 

    Args:
        location (list[float]): [latitude of the center of the landscape, longitude of the center of the landscape]
        start_date (str): the start date  
        end_date (str): _description_
        width (float, optional): _description_. Defaults to 0.01.
        len (float, optional): _description_. Defaults to 0.01.
        scale (int, optional): _description_. Defaults to 5.

    Returns:
        img: image as numpy array
    """
    # reversed longitude latitude
    location = (location[1], location[0])
    ee_img, geometry = extract_ee_img(location, width,start_date,end_date , len)
    url = get_url(ee_img, geometry, scale)
    img = extract_np_from_url(url)
    
    return img

# transform img from numpy to PIL
def transform_ee_img(img, min = 0, max=0.3):
    """Transform an img from numpy to PIL

    Args:
        img (numpy array): the original image as a numpy array 
        min (int, optional): _description_. Defaults to 0.
        max (float, optional): _description_. Defaults to 0.3.

    Returns:
        img_test: a PIL image 
    """
    img=np.minimum(img*255/max,np.ones(img.shape)*255)
    img=np.uint8((np.rint(img)).astype(int))
    return img