Skip to content

Project Canopy's Proposed Changes (v1) #2

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
davidanagy wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Project Canopy's Proposed Changes (v1) #2

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
135 changes: 94 additions & 41 deletions download_sen12.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,9 @@
import time
import os
import json
import time
from argparse import ArgumentParser
from utils import clipToROI, exportImageCollectionToGCS, exportImageToGCS, sentinel2CloudScore, calcCloudCoverage
from utils import clipToROI, exportImageCollectionToGCS, exportImageToGCS, sentinel2CloudScore, calcCloudCoverage, inject_B10, sentinel2ProjectShadows, computeQualityScore, mergeCollection
from utils import GEETaskManager

from gevent.fileobject import FileObjectThread
Expand All @@ -16,9 +17,9 @@ def makeFilterList(sensor):
def _build_filters(filter_list):
filters = []
for f in filter_list:
key = f.keys()[0]
op = f.values()[0].keys()[0]
val = f.values()[0].values()[0]
key = list(f.keys())[0]
op = list(list(f.values())[0].keys())[0]
val = list(list(f.values())[0].values())[0]
filters.append(getattr(ee.Filter, op)(key, val))

return filters
Expand All @@ -33,93 +34,140 @@ def _build_filters(filter_list):

def makeImageCollection(sensor, roi, start_date, end_date, modifiers=[]):
filters_before, filters_after = makeFilterList(sensor)
print(modifiers)

collection = ee.ImageCollection(sensor['name']) \
.filterDate(ee.Date(start_date), ee.Date(end_date)) \
.filterBounds(roi) \
.map( lambda x: clipToROI(x, ee.Geometry(roi)) )

print("size of collection:",collection.size().getInfo())

if filters_before is not None:
collection = collection.filter( filters_before )

if modifiers and len(modifiers) > 0:
for m in modifiers:
print(f'Applying modifier {m}')
collection = collection.map(m)

if filters_after:
collection = collection.filter( filters_after )

return collection.select(sensor['bands'])
return collection

def process_datasource(task_queue, source, sensor, export_to, export_dest):
feature_list = ee.FeatureCollection(source['features_src'])
def process_datasource(source, sensor, export_folder, feature_list = None, pre_mosaic_sort='CLOUDY_PERCENTAGE'):
# feature_list = ee.FeatureCollection(source['features_src'])
feature_list = feature_list.sort(source['sort_by']).toList(feature_list.size())
n_features = feature_list.size().getInfo()

print("{} features have been loaded".format(n_features))

task_list = []
#task_list = []

exports = []

for i in range(1, n_features):
### ERROR? ###
## Originally this was range(1, n_features), but we're pretty sure
## that should be 0 so we changed it.
for i in range(0, n_features):
feature_point = ee.Feature( feature_list.get(i) )

if source['geometry'] == "point":
feature_point = feature_point.buffer(source['size']).bounds()
#if source['geometry'] == "point":
# feature_point = feature_point.buffer(source['size']).bounds()

roi = feature_point.geometry()
roi = roi.coordinates().getInfo()

### should be done outside the for loop ###
if isinstance(source['name'], str):
source['name'] = [source['name']]

### ERROR? ###
## The following conditional should be moved under
## the conditional after it, or else we'll error out
## if sensor doesn't have a "prefix" key.
if isinstance(sensor['prefix'], str):
sensor['prefix'] = [sensor['prefix']]

if 'prefix' in sensor:
filename_parts = sensor['prefix'] + source['name']
else:
filename_parts = source['name']
### end of part that should be done outside the for loop ###

filename = "_".join(source['name'] + [str(i)])
time_stamp = "_".join(time.ctime().split(" ")[1:])
filename = "_".join([str(i + 1)] + source['name'] + [time_stamp])
print("processing ",filename)
dest_path = "/".join(filename_parts + [filename])

export_params = {
'bucket': export_dest,
'bucket': export_folder,
'resolution': source['resolution'],
'filename': filename,
'dest_path': dest_path
}

task_params = {
'action': export_single_feature,
'id': "_".join(filename_parts + [str(i)]), # This must be unique per task, to allow to track retries
'kwargs': {
'roi': roi,
'export_params': export_params,
'type': sensor['type'],
'date_range': {'start_date': source['start_date'], 'end_date': source['end_date']}
}
}

task_queue.add_task(task_params, blocking=True)

def export_single_feature(roi=None, type=None, date_range=None, export_params=None):
modifiers = None
# task_params = {
# 'action': export_single_feature,
# 'id': "_".join(filename_parts + [str(i)]), # This must be unique per task, to allow to track retries
# 'kwargs': {
# 'roi': roi,
# 'export_params': export_params,
# 'sensor': sensor,
# 'date_range': {'start_date': source['start_date'], 'end_date': source['end_date'],
# 'sort_by': pre_mosaic_sort}
# }
# }

# task_queue.add_task(task_params, blocking=True)

export = export_single_feature(
roi=roi,
sensor=sensor,
date_range={'start_date': source['start_date'], 'end_date': source['end_date']},
export_params=export_params,
sort_by=pre_mosaic_sort
)

exports.append(export)

return exports

def export_single_feature(roi=None, sensor=None, date_range=None, export_params=None, sort_by='CLOUDY_PERCENTAGE'):
modifiers = []
if sensor['name'].lower() == "copernicus/s2_sr":
print('Inject B10')
modifiers.append(inject_B10)
if sensor['type'].lower() == "opt":
modifiers = [sentinel2CloudScore, calcCloudCoverage]
#print(sensor['type'])
modifiers += [sentinel2CloudScore, calcCloudCoverage, sentinel2ProjectShadows, computeQualityScore]
print(modifiers)


#print('Modifiers:', modifiers)
roi_ee = ee.Geometry.Polygon(roi[0])
image_collection = makeImageCollection(sensor, roi_ee, date_range['start_date'], date_range['end_date'], modifiers=modifiers)
img = ee.Image(image_collection.mosaic())
## sort was not in the original version
image_collection = image_collection.sort(sort_by)
## below line was in the original verson;
## changing to the JS version
## img = image_collection.mosaic().clip(roi_ee)
cloudFree = mergeCollection(image_collection).clip(roi_ee)
cloudFree = cloudFree.reproject('EPSG:4326', None, 10)
### Do we need to mosaic it now???
# print('cloudFree info:', cloudFree.getInfo())
#print('Mosaic type:', type(img))

new_params = export_params.copy()
new_params['img'] = img
new_params['img'] = cloudFree
new_params['roi'] = roi

new_params['sensor_name'] = sensor['name'].lower()

return exportImageToGCS(**new_params)

def _serialise_task_log(task_log):
for k,v in task_log.iteritems():
for k,v in task_log.items():
task_log[k]['task_def']['action'] = "export_single_feature"

return task_log
Expand All @@ -128,7 +176,7 @@ def load_task_log(filename='task_log.json'):
with open(filename, 'r') as f:
task_log = json.load(f)

for k, v in task_log.iteritems():
for k, v in task_log.items():
task_log[k]['task_def']['action'] = globals()[task_log[k]['task_def']['action']]

return task_log
Expand All @@ -142,12 +190,16 @@ def monitor_tasks(task_log):

f_raw.close()

def load_config(path):
with open(path, 'r') as stream:
try:
return yaml.load(stream)
except yaml.YAMLError as exc:
print(exc)
# def load_config(path):
# with open(path, 'r') as stream:
# try:
# return yaml.load(stream)
# except yaml.YAMLError as exc:
# print(exc)

def load_config(config_file):
stream = open(config_file, 'r')
return yaml.load(stream)

if __name__ == "__main__":
parser = ArgumentParser()
Expand All @@ -167,10 +219,11 @@ def load_config(path):
task_log = load_task_log(filename='task_log.json')
task_queue.set_task_log(task_log)

pre_mosaic_sort = config['pre_mosaic_sort']
for data_list in config['data_list']:
for sensor_idx in data_list['sensors']:
sensor = config['sensors'][sensor_idx]
tasks = process_datasource(task_queue, data_list, sensor, config['export_to'], config['export_dest'])
tasks = process_datasource(task_queue, data_list, sensor, config['export_to'], config['export_dest'], pre_mosaic_sort)

print("Waiting for completion...")
task_queue.wait_till_done()