Process creating NcML file aggregating netCDF files

emu/processes/__init__.py

@@ -16,8 +16,10 @@
 from .wps_ncmeta import NCMeta
 from .wps_nonpyid import NonPyID
 from .wps_dry_run import SimpleDryRun
+from .wps_ncml import NcMLAgg
 from .wps_translation import Translation
 
+
 processes = [
     UltimateQuestion(),
     Sleep(),
@@ -37,5 +39,6 @@
     NCMeta(),
     NonPyID(),
     SimpleDryRun(),
+    NcMLAgg(),
     Translation(),
 ]

emu/processes/wps_nc_to_dap.py

@@ -0,0 +1,53 @@
+from pywps import Process
+from pywps import ComplexInput, ComplexOutput, FORMATS
+from pywps.app.Common import Metadata
+from pywps import configuration
+import logging
+
+
+LOGGER = logging.getLogger("PYWPS")
+
+# This needs some more thoughts before going live to decide how the WPS server should be configured to communicate
+# with the DAP server.
+
+
+class NcToDap(Process):
+    def __init__(self):
+        inputs = [
+            ComplexInput('resource', "NetCDF file",
+                         abstract="Link to NetCDF or NcML file on this server",
+                         supported_formats=[FORMATS.NETCDF, ],  # FORMATS.NCML], to become available in PyWPS 4.2.5
+                         min_occurs=1,
+                         max_occurs=1)
+        ]
+        outputs = [
+            ComplexOutput('dap', 'DAP url',
+                          as_reference=True,
+                          supported_formats=[FORMATS.DODS]),
+        ]
+
+        super(NcToDap, self).__init__(
+            self._handler,
+            identifier='nc_to_dap',
+            title="Convert file URL to DAP URL",
+            abstract="Return Data Access Protocol link to a netCDF or NcML file.",
+            version="1",
+            metadata=[
+                Metadata('User Guide', 'http://emu.readthedocs.io/en/latest/'),
+            ],
+            inputs=inputs,
+            outputs=outputs,
+            store_supported=True,
+            status_supported=True)
+
+    @staticmethod
+    def _handler(self, request, response):
+        url = request.inputs['resource'][0].url
+
+        # Write response
+        file_server = configuration.CONFIG.get("server", "outputurl")
+        dap_server = configuration.CONFIG.get("dap", "outputurl")
+
+        response.outputs["dap"].url = url.replace(file_server, dap_server)
+
+        return response

emu/processes/wps_ncml.py

@@ -0,0 +1,68 @@
+import os
+import xarray.tests.test_dataset as td
+from pywps import Process
+from pywps import ComplexOutput, FORMATS
+from pywps.app.Common import Metadata
+import logging
+
+
+LOGGER = logging.getLogger("PYWPS")
+
+
+class NcMLAgg(Process):
+    def __init__(self):
+        inputs = []
+        outputs = [
+            ComplexOutput('d1', 'NetCDF file 1',
+                          as_reference=True,
+                          supported_formats=[FORMATS.NETCDF]),
+            ComplexOutput('d2', 'NetCDF file 2',
+                          as_reference=True,
+                          supported_formats=[FORMATS.NETCDF]),
+            ComplexOutput('ncml', 'NcML aggregation',
+                          as_reference=True,
+                          supported_formats=[FORMATS.DODS]),  # FORMATS.NCML To become available in PyWPS 4.2.5
+        ]
+
+        super(NcMLAgg, self).__init__(
+            self._handler,
+            identifier='ncml',
+            title="Test NcML THREDDS capability",
+            abstract="Return links to an NcML file aggregating netCDF files with moving time units.",
+            version="1",
+            metadata=[
+                Metadata('User Guide', 'http://emu.readthedocs.io/en/latest/'),
+            ],
+            inputs=inputs,
+            outputs=outputs,
+            store_supported=True,
+            status_supported=True)
+
+    def _handler(self, request, response):
+
+        # Create test datasets
+        d1, d2, _ = td.create_append_test_data()
+
+        # Save datasets to disk
+        d1fn = os.path.join(self.workdir, "d1.nc")
+        d2fn = os.path.join(self.workdir, "d2.nc")
+
+        d1.to_netcdf(d1fn)
+        d2.to_netcdf(d2fn)
+
+        # Create NcML aggregation
+        ncml = """
+        <netcdf xmlns="http://www.unidata.ucar.edu/namespaces/netcdf/ncml-2.2">
+            <aggregation dimName="time" type="joinExisting">
+                <scan location="." suffix=".nc" subdirs="false"/>
+            </aggregation>
+        </netcdf>
+        """
+
+        # Write response
+        response.outputs["d1"].file = d1fn
+        response.outputs["d2"].file = d2fn
+
+        response.outputs['ncml'].data = ncml
+
+        return response

requirements.txt

@@ -4,3 +4,4 @@ click
 psutil
 defusedxml
 geomet
+netCDF4

setup.cfg

@@ -19,18 +19,20 @@ search = Version="{current_version}"
 replace = Version="{new_version}"
 
 [tool:pytest]
-addopts = 
+addopts =
 	--strict
 	--tb=native
 python_files = test_*.py
-markers = 
+markers =
 	online: mark test to need internet connection
 	slow: mark test to be slow
+	flaky: mark test to be flaky
+	network: mark test to need network
 
 [flake8]
 ignore = F401,E402
 max-line-length = 120
-exclude = 
+exclude =
 	.git,
 	__pycache__,
 	docs/source/conf.py,

tests/test_wps_caps.py

@@ -23,6 +23,7 @@ def test_wps_caps():
         'multiple_outputs',
         'nap',
         'ncmeta',
+        'ncml',
         'non.py-id',
         'output_formats',
         'poly_centroid',

tests/test_wps_nc_to_dap.py

@@ -0,0 +1,28 @@
+import pytest
+
+from pywps import Service
+from pywps.tests import assert_response_success
+
+from .common import client_for, get_output
+from emu.processes.wps_nc_to_dap import NcToDap
+from emu.processes.wps_output_formats import OutputFormats
+
+
+@pytest.mark.skip
+def test_wps_nc_to_dap():
+    # Create netCDF file on server
+    client = client_for(Service(processes=[OutputFormats()]))
+    resp = client.get(
+        service='WPS', request='Execute', version='1.0.0',
+        identifier='output_formats',)
+    nc_url = get_output(resp.xml)['netcdf']
+
+    # Convert to DAP link
+    client = client_for(Service(processes=[NcToDap()]))
+    datainputs = "resource=@xlink:href={0}".format(nc_url)
+    resp = client.get(
+        service='wps', request='execute', version='1.0.0',
+        identifier='nc_to_dap',
+        datainputs=datainputs)
+    assert_response_success(resp)
+    assert 'dodsC' in get_output(resp.xml)['dap']

tests/test_wps_ncml.py

@@ -0,0 +1,25 @@
+import pytest
+
+from pywps import Service
+from pywps.tests import assert_response_success
+from emu.processes.wps_ncml import NcMLAgg
+from .common import client_for, CFG_FILE
+from owslib.wps import WPSExecution
+
+
+@pytest.mark.online
+def test_wps_ncml():
+
+    client = client_for(Service(processes=[NcMLAgg()], cfgfiles=CFG_FILE))
+
+    resp = client.get(
+        service='wps', request='execute', version='1.0.0',
+        identifier='ncml')
+
+    assert_response_success(resp)
+
+    ex = WPSExecution()
+    ex.parseResponse(resp.xml)
+    d1, d2, d3 = ex.processOutputs
+    ncml = d3.retrieveData()
+    assert ncml.strip().startswith("<")

tests/test_wps_output_formats.py

@@ -1,12 +1,12 @@
 from pywps import Service
 from pywps.tests import assert_response_success
 
-from .common import client_for
+from .common import client_for, CFG_FILE
 from emu.processes.wps_output_formats import OutputFormats
 
 
 def test_wps_output_formats():
-    client = client_for(Service(processes=[OutputFormats()]))
+    client = client_for(Service(processes=[OutputFormats()], cfgfiles=CFG_FILE))
     resp = client.get(
         service='WPS', request='Execute', version='1.0.0',
         identifier='output_formats',)