WFP-01-03-01 Data transformation application - COPERNICUS Vegetation Indicators - Aggregations and AnomaliesΒΆ

This Jupyter Notebook to query the catalog for a COPERNICUS product (LAI/FAPAR), creates a Web Processing Service (WPS) request invoking the data transformation application that was deployed in the Deploy step, monitors the WPS request execution and finally retrieves the data transformation execution results

  • First do the imports of the Python libraries required.
In [3]:

import os
import owslib
from owslib.wps import monitorExecution
from owslib.wps import WebProcessingService
import lxml.etree as etree
import cioppy

from shapely.wkt import loads

import getpass

from nbconvert.preprocessors import ExecutePreprocessor, CellExecutionError
import nbformat as nbf
  • Read the data pipeline configuration information:
In [4]:

%store -r

nb_config = os.path.join('../operations', 'configuration.ipynb')

nb = nbf.read(nb_config, 4)

exec(nb['cells'][1]['source']) in globals(), locals()

app = dict([('artifact_id', app_artifact_id.replace('ewf-', '')),
            ('version', app_version),
            ('repository', repository),
            ('community', community)])

app_process_id = '%s_%s_%s_%s' % (app['community'].replace('-', '_'), app['artifact_id'].replace('-', '_'), app['artifact_id'].replace('-', '_'), app['version'].replace('.', '_'))

In [5]:

print(app_process_id)

ec_better_wfp_01_03_01_wfp_01_03_01_1_5
  • Define the search parameters: the catalog series OpenSearch endpoint, the time of interest and the area of interest
In [4]:

series = 'https://catalog.terradue.com/cgls/description'

start_date = '2015-01-01T00:00:00Z'
stop_date = '2015-01-31T23:59:59Z'
count = 366

geom = 'MULTIPOLYGON (((19.564118379527372 -14.910538401469323,27.210602754527372 -14.910538401469323,27.210602754527372 -21.09243333786736,19.564118379527372 -21.09243333786736,19.564118379527372 -14.910538401469323)), ((12.9415 13.7579, 14.6731 13.7579, 14.6731 12.0093, 12.9415 12.0093, 12.9415 13.7579)))'

Limit the test to the AOIs:

In [5]:

wkt = loads(geom)[0].wkt

print wkt

POLYGON ((19.56411837952737 -14.91053840146932, 27.21060275452737 -14.91053840146932, 27.21060275452737 -21.09243333786736, 19.56411837952737 -21.09243333786736, 19.56411837952737 -14.91053840146932))
  • Search for COPERNICUS products
In [6]:

search_params = dict([('start', start_date),
                      ('stop', stop_date),
                      ('count', count)])


In [7]:

ciop = cioppy.Cioppy()

search = ciop.search(end_point=series,
                     params=search_params,
                     output_fields='self,enclosure,identifier',
                     model='GeoTime')
  • List the COPERNICUS products found
In [8]:

for index, elem in enumerate(search):
    print(index, elem['identifier'])

(0, 'fapar_v2_1km_FAPAR-RT6_201501310000_GLOBE_PROBAV_V2.0.2')
(1, 'lai_v2_1km_LAI-RT6_201501310000_GLOBE_PROBAV_V2.0.2')
(2, 'fapar_v2_1km_FAPAR-RT6_201501200000_GLOBE_PROBAV_V2.0.2')
(3, 'lai_v2_1km_LAI-RT6_201501200000_GLOBE_PROBAV_V2.0.2')
(4, 'fapar_v2_1km_FAPAR-RT6_201501100000_GLOBE_PROBAV_V2.0.2')
(5, 'lai_v2_1km_LAI-RT6_201501100000_GLOBE_PROBAV_V2.0.2')
  • Connect to WPS server and do a GetCapabilities WPS request and list the process:
In [9]:

wps_url = 'https://ec-better-apps-deployer.terradue.com/zoo-bin/zoo_loader.cgi'
wps = WebProcessingService(wps_url, verbose=False, skip_caps=True)

wps.getcapabilities()

#process_id = 'ec_better_wfp_01_03_02_wfp_01_03_02_1_6'

process = wps.describeprocess(app_process_id)

for input in process.dataInputs:
    print(input.identifier)

source
N_1
N_3
N_6
N_9
N_12
N_15
N_18
N_27
N_36
regionOfInterest
nameOfRegion
_T2Username

In [10]:

N_1 = 'False'
N_3 = 'True'
N_6 = 'False'
N_9 = 'False'
N_12 = 'False'
N_15 = 'False'
N_18 = 'False'
N_27 = 'False'
N_36 = 'False'

In [11]:

inputs = []
for catalogue_url in search:
    inputs.append(('source', catalogue_url['self']))
inputs_2 = [('N_1', N_1), ('N_3', N_3), ('N_6', N_6), ('N_9', N_9), ('N_12', N_12), ('N_15', N_15),('N_18', N_18),('N_27', N_27), ('N_36', N_36),
           ('regionOfInterest', wkt),
           ('nameOfRegion', 'Sudan'),
           ('_T2Username', 'rirr')]
inputs += inputs_2
print(inputs)

[('source', 'https://catalog.terradue.com/cgls/search?format=atom&uid=fapar_v2_1km_FAPAR-RT6_201703310000_GLOBE_PROBAV_V2.0.2'), ('source', 'https://catalog.terradue.com/cgls/search?format=atom&uid=lai_v2_1km_LAI-RT6_201703310000_GLOBE_PROBAV_V2.0.2'), ('source', 'https://catalog.terradue.com/cgls/search?format=atom&uid=fapar_v2_1km_FAPAR-RT6_201703200000_GLOBE_PROBAV_V2.0.2'), ('source', 'https://catalog.terradue.com/cgls/search?format=atom&uid=lai_v2_1km_LAI-RT6_201703200000_GLOBE_PROBAV_V2.0.2'), ('source', 'https://catalog.terradue.com/cgls/search?format=atom&uid=fapar_v2_1km_FAPAR-RT6_201703100000_GLOBE_PROBAV_V2.0.2'), ('source', 'https://catalog.terradue.com/cgls/search?format=atom&uid=lai_v2_1km_LAI-RT6_201703100000_GLOBE_PROBAV_V2.0.2'), ('source', 'https://catalog.terradue.com/cgls/search?format=atom&uid=fapar_v2_1km_FAPAR-RT6_201702280000_GLOBE_PROBAV_V2.0.2'), ('source', 'https://catalog.terradue.com/cgls/search?format=atom&uid=lai_v2_1km_LAI-RT6_201702280000_GLOBE_PROBAV_V2.0.2'), ('source', 'https://catalog.terradue.com/cgls/search?format=atom&uid=fapar_v2_1km_FAPAR-RT6_201702200000_GLOBE_PROBAV_V2.0.2'), ('source', 'https://catalog.terradue.com/cgls/search?format=atom&uid=lai_v2_1km_LAI-RT6_201702200000_GLOBE_PROBAV_V2.0.2'), ('source', 'https://catalog.terradue.com/cgls/search?format=atom&uid=fapar_v2_1km_FAPAR-RT6_201702100000_GLOBE_PROBAV_V2.0.2'), ('source', 'https://catalog.terradue.com/cgls/search?format=atom&uid=lai_v2_1km_LAI-RT6_201702100000_GLOBE_PROBAV_V2.0.2'), ('source', 'https://catalog.terradue.com/cgls/search?format=atom&uid=fapar_v2_1km_FAPAR-RT6_201701310000_GLOBE_PROBAV_V2.0.2'), ('source', 'https://catalog.terradue.com/cgls/search?format=atom&uid=lai_v2_1km_LAI-RT6_201701310000_GLOBE_PROBAV_V2.0.2'), ('source', 'https://catalog.terradue.com/cgls/search?format=atom&uid=fapar_v2_1km_FAPAR-RT6_201701200000_GLOBE_PROBAV_V2.0.2'), ('source', 'https://catalog.terradue.com/cgls/search?format=atom&uid=lai_v2_1km_LAI-RT6_201701200000_GLOBE_PROBAV_V2.0.2'), ('source', 'https://catalog.terradue.com/cgls/search?format=atom&uid=fapar_v2_1km_FAPAR-RT6_201701100000_GLOBE_PROBAV_V2.0.2'), ('source', 'https://catalog.terradue.com/cgls/search?format=atom&uid=lai_v2_1km_LAI-RT6_201701100000_GLOBE_PROBAV_V2.0.2'), ('N_1', 'False'), ('N_3', 'True'), ('N_6', 'False'), ('N_9', 'False'), ('N_12', 'False'), ('N_15', 'False'), ('N_18', 'False'), ('N_27', 'False'), ('N_36', 'False'), ('regionOfInterest', 'POLYGON ((26.832 9.5136, 28.6843 9.5136, 28.6843 7.8009, 26.832 7.8009, 26.832 9.5136))'), ('nameOfRegion', 'Sudan'), ('_T2Username', 'rirr')]

In [12]:

app_deployed = False

for index, elem in enumerate(wps.processes):
    if elem.identifier == app_process_id:
        app_deployed = True

if app_deployed:
    print 'Process %s deployed' % app_process_id
else:
    raise Exception('Process %s not deployed' % app_process_id)

Process ec_better_wfp_01_03_01_wfp_01_03_01_1_5 deployed
  • Select the process and print the title and abstract after having submited a WPS DescribeProcess request
In [13]:

process = wps.describeprocess(app_process_id)

print process.title

print process.abstract

COPERNICUS Vegetation Indicators - Aggregations
COPERNICUS Vegetation Indicators - Aggregations
  • List the WPS process inputs:
In [14]:

for data_input in process.dataInputs:
    print data_input.identifier

source
N_1
N_3
N_6
N_9
N_12
N_15
N_18
N_27
N_36
regionOfInterest
nameOfRegion
_T2Username
  • Submit the Execute WPS request:
In [15]:

execution = owslib.wps.WPSExecution(url=wps.url)

execution_request = execution.buildRequest(app_process_id,
                                           inputs,
                                           output=[('result_osd', False)])

execution_response = execution.submitRequest(etree.tostring(execution_request))

execution.parseResponse(execution_response)
print(execution.statusLocation)

http://ec-better-apps-deployer.terradue.com/zoo-bin/zoo_loader.cgi?request=Execute&service=WPS&version=1.0.0&Identifier=GetStatus&DataInputs=sid=624b9a8a-2a2c-11e9-929b-0242ac11000f&RawDataOutput=Result
  • Monitor the request:
In [16]:

execution.statusLocation

Out[16]:

'http://ec-better-apps-deployer.terradue.com/zoo-bin/zoo_loader.cgi?request=Execute&service=WPS&version=1.0.0&Identifier=GetStatus&DataInputs=sid=624b9a8a-2a2c-11e9-929b-0242ac11000f&RawDataOutput=Result'

In [17]:

monitorExecution(execution)
  • Check the outcome of the processing request
In [18]:

if not execution.isSucceded():
    raise Exception('Processing failed')
  • Search for the results produced
In [19]:

results_osd = execution.processOutputs[0].reference

print results_osd

https://recast.terradue.com/t2api/describe/rirr/_results/workflows/ec_better_wfp_01_03_01_wfp_01_03_01_1_5/run/624b9a8a-2a2c-11e9-929b-0242ac11000f/0003516-181221095105003-oozie-oozi-W

In [21]:

from geopandas import GeoDataFrame
search_results = GeoDataFrame(ciop.search(end_point=results_osd,
                         params=[],
                         output_fields='title,enclosure',
                         model='GeoTime',
                            timeout=50000))

search_results

Out[21]:
enclosure title
0 https://store.terradue.com/rirr/_results/workf... Output FAPAR_Sudan_N3_averages_2017-01-10_2017...
1 https://store.terradue.com/rirr/_results/workf... Output FAPAR_Sudan_N3_averages_2017-02-10_2017...
2 https://store.terradue.com/rirr/_results/workf... Output FAPAR_Sudan_N3_averages_2017-03-10_2017...
3 https://store.terradue.com/rirr/_results/workf... Output FAPAR_Sudan_N3_maxvalues_2017-01-10_201...
4 https://store.terradue.com/rirr/_results/workf... Output FAPAR_Sudan_N3_maxvalues_2017-02-10_201...
5 https://store.terradue.com/rirr/_results/workf... Output FAPAR_Sudan_N3_maxvalues_2017-03-10_201...
6 https://store.terradue.com/rirr/_results/workf... Output LAI_Sudan_N3_averages_2017-01-10_2017-0...
7 https://store.terradue.com/rirr/_results/workf... Output LAI_Sudan_N3_averages_2017-02-10_2017-0...
8 https://store.terradue.com/rirr/_results/workf... Output LAI_Sudan_N3_averages_2017-03-10_2017-0...
9 https://store.terradue.com/rirr/_results/workf... Output LAI_Sudan_N3_maxvalues_2017-01-10_2017-...
10 https://store.terradue.com/rirr/_results/workf... Output LAI_Sudan_N3_maxvalues_2017-02-10_2017-...
11 https://store.terradue.com/rirr/_results/workf... Output LAI_Sudan_N3_maxvalues_2017-03-10_2017-...
12 https://store.terradue.com/rirr/_results/workf... result.ipynb 
In [ ]:

recast_process_id = 'dataPublication'
recast_wps_url = 'https://recast.terradue.com/t2api/ows'

wps = WebProcessingService(recast_wps_url,
                           verbose=False,
                           skip_caps=False)

recast_inputs = [('items', results_osd),
                  ('index', data_pipeline),
                  ('_T2ApiKey', data_pipeline_api_key),
                  ('_T2Username', data_pipeline)]

recast_execution = wps.execute(recast_process_id,
                               recast_inputs,
                               output = [('result_osd', True)])


monitorExecution(recast_execution, sleepSecs=60)

etree.fromstring(recast_execution.processOutputs[0].data[0]).xpath('./@href')[0]