Process creating NcML file aggregating netCDF files

emu/processes/__init__.py

@@ -16,6 +16,8 @@
 from .wps_ncmeta import NCMeta
 from .wps_nonpyid import NonPyID
 from .wps_dry_run import SimpleDryRun
+from .wps_ncml import NcMLAgg
+from .wps_nc_to_dap import NcToDap
 
 processes = [
     UltimateQuestion(),
@@ -36,4 +38,6 @@
     NCMeta(),
     NonPyID(),
     SimpleDryRun(),
+    NcMLAgg(),
+    NcToDap(),
 ]

emu/processes/wps_nc_to_dap.py

@@ -0,0 +1,51 @@
+import os
+
+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")
+
+
+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],
+                         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)
+
+    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,71 @@
+import os
+
+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.NCML, FORMATS.DODS]),
+        ]
+
+        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):
+        import xarray.tests.test_dataset as td
+
+        # 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
+
+        with open(os.path.join(self.workdir, 'agg.ncml'), "w") as fp:
+            response.outputs['ncml'].file = fp.name
+            fp.write(ncml)
+
+        return response

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

@@ -22,7 +22,9 @@ def test_wps_caps():
         'inout',
         'multiple_outputs',
         'nap',
+        'nc_to_dap',
         'ncmeta',
+        'ncml',
         'non.py-id',
         'output_formats',
         'poly_centroid',

tests/test_wps_nc_to_dap.py

@@ -0,0 +1,30 @@
+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.online
+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,18 @@
+import pytest
+
+from pywps import Service
+from pywps.tests import assert_response_success
+
+from .common import client_for
+from emu.processes.wps_ncml import NcMLAgg
+
+
+@pytest.mark.online
+def test_wps_ncml():
+    client = client_for(Service(processes=[NcMLAgg()]))
+
+    resp = client.get(
+        service='wps', request='execute', version='1.0.0',
+        identifier='ncml')
+    print(resp.response)
+    assert_response_success(resp)