Test a data stream entryΒΆ

In [32]:

import owslib
from owslib.wps import monitorExecution
import uuid
from owslib.wps import WebProcessingService
import sys
import os
import lxml.etree as etree
import requests
import cioppy
ciop = cioppy.Cioppy()
import shapely
import dateutil.parser
from shapely.wkt import loads
import pandas as pd
import geopandas as gp
from datetime import datetime, timedelta
from io import BytesIO
from zipfile import ZipFile
import matplotlib.pyplot as plt
import gdal
import numpy as np
import requests

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

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

nb = nbf.read(nb_config, 4)

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

In [34]:

series = 'https://catalog.terradue.com/%s/series/source-queue/search' % data_pipeline

In [35]:

series

Out[35]:

'https://catalog.terradue.com/better-wfp-00003/series/source-queue/search'

In [36]:

print data_pipeline

better-wfp-00003

In [46]:

search_params = dict([('count', '50')])

In [47]:

update = '2018-08-07T08:00:00Z/2018-10-08T09:00:00Z'

In [48]:

search_params

Out[48]:

{'count': '50'}

In [49]:

entry_search = ciop.search(end_point=series,
                           params=search_params,
                           output_fields='self',
                           model='EOP')

In [50]:

for item in entry_search:
    print item

{'self': 'https://catalog.terradue.com//better-wfp-00003/series/source-queue/search?format=atomeop&uid=S2A_MSIL2A_20180920T095021_N0208_R079_T33SWD_20180920T130438'}
{'self': 'https://catalog.terradue.com//better-wfp-00003/series/source-queue/search?format=atomeop&uid=S2A_MSIL2A_20180920T095021_N0208_R079_T33SXC_20180920T130438'}

In [16]:

stream = entry_search[0]['self']

In [17]:

stream

Out[17]:

'https://catalog.terradue.com//better-wfp-00003/series/source-queue/search?format=atomeop&uid=S2B_MSIL2A_20181022T094029_N0209_R036_T33SVB_20181022T143356'
  • Read the stream OWS Context document
In [18]:

root = etree.fromstring(requests.get(stream).content)
  • Get the Web Process Service access point:
In [19]:

ns = {'a':'http://www.w3.org/2005/Atom',
      'b':'http://www.opengis.net/owc/1.0',
      'c':'http://www.opengis.net/wps/1.0.0',
      'd':'http://www.opengis.net/ows/1.1'}

In [20]:

wps_url = root.xpath('/a:feed/a:entry/b:offering/b:operation[@code="Execute"]',
                                 namespaces=ns)[0].attrib['href']
  • Get the process identifier:
In [21]:

process_id = root.xpath('/a:feed/a:entry/b:offering/b:operation[@code="Execute"]/b:request/c:Execute/d:Identifier',
                                 namespaces=ns)[0].text
  • Check that the process is deployed
In [22]:

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

In [23]:

wps.getcapabilities()

In [24]:

deployed = False

for index, elem in enumerate(wps.processes):
    if elem.identifier == process_id:
        deployed = True
        break

if not deployed:
    raise Exception('Process %s not deployed' % process_id)
  • Print the parameters
In [25]:

identifiers = root.xpath('/a:feed/a:entry/b:offering/b:operation[@code="Execute"]/b:request/c:Execute/c:DataInputs/c:Input/d:Identifier',
                                 namespaces=ns)

values = root.xpath('/a:feed/a:entry/b:offering/b:operation[@code="Execute"]/b:request/c:Execute/c:DataInputs/c:Input/c:Data/c:LiteralData',
                                 namespaces=ns)

params = dict()

for index, elem in enumerate(identifiers):

    params[elem.text] = values[index].text


params

Out[25]:

{'_T2Username': 'better-wfp-00003',
 'flag_expr': 'saturated_l1a_B4 or scl_water',
 'resolution': '10',
 'source': 'https://catalog.terradue.com/sentinel2/search?format=atom&uid=S2B_MSIL2A_20181022T094029_N0209_R036_T33SVB_20181022T143356\n',
 'wkt': '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))'}
  • Submit the WPS request:
In [26]:

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

In [27]:

execution_request = root.xpath('/a:feed/a:entry/b:offering/b:operation[@code="Execute"]/b:request/c:Execute',
                                 namespaces=ns)[0]


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

In [28]:

execution.parseResponse(execution_response)
  • Status location of the submitted process:
In [29]:

execution.statusLocation

Out[29]:

'http://ec-better-apps-deployer.terradue.com/zoo-bin/zoo_loader.cgi?request=Execute&service=WPS&version=1.0.0&Identifier=GetStatus&DataInputs=sid=f29a2dd4-d8d5-11e8-9649-0242ac11000f&RawDataOutput=Result'
  • Monitor the execution
In [30]:

monitorExecution(execution)

In [31]:

print execution.isSucceded()

True
  • Check the output by querying the OpenSearch access points for the results:
In [79]:

for output in execution.processOutputs:
    print(output.identifier)

result_osd
QUOTATION

In [80]:

results_osd = execution.processOutputs[0].reference

In [81]:

results_osd

Out[81]:

'https://recast.terradue.com/t2api/describe/better-wfp-00003/_results/workflows/ec_better_ewf_wfp_01_02_01_ewf_wfp_01_02_01_0_2/run/38b2a210-d6c4-11e8-ac00-0242ac11000f/0021696-180330140554685-oozie-oozi-W'

In [82]:

search_results = ciop.search(end_point=results_osd,
                         params=[],
                         output_fields='identifier,enclosure',
                         model='GeoTime')

In [83]:

for index, elem in enumerate(search_results):

    print(elem['enclosure'])

https://store.terradue.com/better-wfp-00003/_results/workflows/ec_better_ewf_wfp_01_02_01_ewf_wfp_01_02_01_0_2/run/38b2a210-d6c4-11e8-ac00-0242ac11000f/0021696-180330140554685-oozie-oozi-W/1f01fe73-f563-4061-b376-e8017309f2c2/result.ipynb
https://store.terradue.com/better-wfp-00003/_results/workflows/ec_better_ewf_wfp_01_02_01_ewf_wfp_01_02_01_0_2/run/38b2a210-d6c4-11e8-ac00-0242ac11000f/0021696-180330140554685-oozie-oozi-W/1f01fe73-f563-4061-b376-e8017309f2c2/stage-in.ipynb
https://store.terradue.com/better-wfp-00003/_results/workflows/ec_better_ewf_wfp_01_02_01_ewf_wfp_01_02_01_0_2/run/38b2a210-d6c4-11e8-ac00-0242ac11000f/0021696-180330140554685-oozie-oozi-W/1f01fe73-f563-4061-b376-e8017309f2c2/S2A_MSIL2A_20181020T095031_N0209_R079_T33SVB_20181020T113743_VI_BOA.gz
  • Download the results to the local filesystem
In [84]:

headers = {'Authorization': 'Bearer %s' % access_token,
           'User-Agent': 'curl/t2Client'}

In [85]:

for index, elem in enumerate(search_results):

    r = requests.get(elem['enclosure'], headers=headers)

    filename = elem['enclosure'][elem['enclosure'].rfind("/")+1:]

    print ('Download %s as %s' % (elem['enclosure'], filename))

    open(filename, 'wb').write(r.content)

Download https://store.terradue.com/better-wfp-00003/_results/workflows/ec_better_ewf_wfp_01_02_01_ewf_wfp_01_02_01_0_2/run/38b2a210-d6c4-11e8-ac00-0242ac11000f/0021696-180330140554685-oozie-oozi-W/1f01fe73-f563-4061-b376-e8017309f2c2/result.ipynb as result.ipynb
Download https://store.terradue.com/better-wfp-00003/_results/workflows/ec_better_ewf_wfp_01_02_01_ewf_wfp_01_02_01_0_2/run/38b2a210-d6c4-11e8-ac00-0242ac11000f/0021696-180330140554685-oozie-oozi-W/1f01fe73-f563-4061-b376-e8017309f2c2/stage-in.ipynb as stage-in.ipynb
Download https://store.terradue.com/better-wfp-00003/_results/workflows/ec_better_ewf_wfp_01_02_01_ewf_wfp_01_02_01_0_2/run/38b2a210-d6c4-11e8-ac00-0242ac11000f/0021696-180330140554685-oozie-oozi-W/1f01fe73-f563-4061-b376-e8017309f2c2/S2A_MSIL2A_20181020T095031_N0209_R079_T33SVB_20181020T113743_VI_BOA.gz as S2A_MSIL2A_20181020T095031_N0209_R079_T33SVB_20181020T113743_VI_BOA.gz