Test the data transformation application

This Jupyter Notebook to query the catalog for a Sentinel-1 GRD product, 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

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:

%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),
            ('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('.', '_'))

• Define the search parameters: the catalog series OpenSearch endpoint, the time of interest and the area of interest

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

start_date = '2018-10-10T00:00:00Z'
stop_date = '2018-10-20T00:00:00Z'

s2_prd_type = 'S2MSI2A'

geom = 'POLYGON((14.9997721265064 38.0212298555682,15.0339421243074 38.1469295398089,15.076722011609 38.2943830685215,15.1196939590664 38.4418167678655,15.1628416993738 38.5892613166416,15.2061510925436 38.7366798078584,15.2388261265461 38.8477042183055,16.2649601148006 38.8421486728812,16.2478654087227 37.8528295032852,14.9997726359092 37.8594419593273,14.9997721265064 38.0212298555682))'

Limit the test to one of the AOIs:

wkt = geom

print wkt

• Search for Sentinel-1 GRD products in one of the polygons of the area of interest

search_params = dict([('geom', wkt),
                      ('start', start_date),
                      ('stop', stop_date),
                      ('pt', s2_prd_type)])

ciop = cioppy.Cioppy()

search = ciop.search(end_point=series,
                     params=search_params,
                     output_fields='self,enclosure,identifier',
                     model='GeoTime')

print app_process_id

• List the Sentinel-1 GRD products found

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

• Select the Sentinel-1 GRD product to process

s2_index = 0

• Connect to the WPS server

wps_url = '%s/zoo-bin/zoo_loader.cgi' % apps_deployer

wps = WebProcessingService(wps_url,
                           verbose=False,
                           skip_caps=True)

• Do a GetCapabilities WPS request and list the process:

wps.getcapabilities()

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)

• Select the process and print the title and abstract after having submited a WPS DescribeProcess request

process = wps.describeprocess(app_process_id)

print process.title

print process.abstract

• List the WPS process inputs:

for data_input in process.dataInputs:
    print data_input.identifier

• Create a Python dictionary with the inputs:

resolution = '10'
percentage_threshold = '20.0'
flag_expr = '( saturated_l1a_B4 or scl_water )'

inputs = [('source', search[s2_index]['self']),
          ('resolution', resolution),
          ('percentage_threshold', percentage_threshold),
          ('flag_expr', flag_expr),
          ('wkt', wkt),
          ('quotation', 'No'),
          ('_T2Username', data_pipeline)]

• Submit the Execute WPS request:

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)

• Monitor the request:

execution.statusLocation

monitorExecution(execution)

• Check the outcome of the processing request

if not execution.isSucceded():
    raise Exception('Processing failed')

• Search for the results produced

results_osd = execution.processOutputs[0].reference

print results_osd

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', datapipeline_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]