Test a data stream entryΒΆ

In [1]:

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 [2]:

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

nb = nbf.read(nb_config, 4)

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

In [3]:



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

In [4]:

print data_pipeline

better-satcen-00001

In [5]:

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

In [6]:

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

In [7]:

stream = entry_search[0]['self']

In [8]:

print stream


https://catalog.terradue.com//better-satcen-00001/series/source-queue/search?format=atomeop&uid=S2A_MSIL2A_20180504T092031_N0207_R093_T34TDK_20180504T112905
  • Read the stream OWS Context document
In [9]:

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

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 [11]:

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

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

In [13]:

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

In [14]:

wps.getcapabilities()

In [15]:

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 [16]:

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
  • Submit the WPS request:
In [17]:

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

In [18]:

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 [19]:

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

execution.statusLocation

Out[20]:

'http://ec-better-apps-deployer.terradue.com/zoo-bin/zoo_loader.cgi?request=Execute&service=WPS&version=1.0.0&Identifier=GetStatus&DataInputs=sid=165f4c3e-35da-11e9-8bdb-0242ac11000f&RawDataOutput=Result'
  • Monitor the execution
In [21]:

monitorExecution(execution)

In [22]:

print execution.isSucceded()

False
  • Check the output by querying the OpenSearch access points for the results:
In [23]:

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

result_osd

In [24]:

results_osd = execution.processOutputs[0].reference

In [25]:

results_osd

In [26]:

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

In [ ]:

for index, elem in enumerate(search_results):

    print(elem['enclosure'])
  • Download the results to the local filesystem
In [ ]:

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

In [ ]:

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)