-
Notifications
You must be signed in to change notification settings - Fork 0
/
api.py
70 lines (43 loc) · 1.92 KB
/
api.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
import os
import sys
from flask import Flask, jsonify, request, Response, redirect
app = Flask(__name__)
app.url_map.strict_slashes = False
# add path to included packages
# these are all stored in the root of the zipped deployment package
root_dir = os.path.dirname(os.path.realpath(__file__))
sys.path.insert(0, root_dir)
from geop import geo_utils, geoprocessing
from serializers import gfw_api
from utilities import util, errors
@app.route("/glad-alerts", methods=['POST'])
@app.route("/glad-alerts/download", methods=['POST'])
def glad_alerts():
geom, area_ha = util.get_shapely_geom()
params = util.validate_glad_params()
if os.environ.get('ENV') == 'test':
glad_raster = os.path.join(root_dir, 'test', 'data', 'afr_all_years_clip.tif')
else:
glad_raster = os.path.join(root_dir, 'data', 'glad.vrt')
if 'download' in request.url_rule.rule:
return download(geom, glad_raster, params)
else:
return stats(geom, glad_raster, params, area_ha)
def stats(geom, glad_raster, params, area_ha):
stats = geoprocessing.count(geom, glad_raster)
hist = util.unpack_glad_histogram(stats, params)
return gfw_api.serialize_glad(hist, area_ha, params['aggregate_by'], params['period'])
def download(geom, glad_raster, params):
masked_data, shifted_affine = geo_utils.mask_geom_on_raster(geom, glad_raster)
# make sure that our AOI covers the raster of interest
if masked_data.any():
# we could do this as a generator, but want to return all download points or fail
rows = [util.filter_rows(row, params) for row in geo_utils.array_to_xyz_rows(masked_data, shifted_affine)]
rows = [n for n in filter(lambda x: x is not False, rows)]
else:
rows = []
s3_url = gfw_api.write_to_s3(rows, params['format'])
return redirect(s3_url, code=302)
@app.errorhandler(errors.Error)
def handle_error(error):
return error.serialize