Bccaqv2 split bbox and grid point

environment.yml

@@ -1,24 +1,23 @@
 name: finch
 channels:
-- birdhouse
-- conda-forge
-- defaults
+  - birdhouse
+  - conda-forge
+  - defaults
 dependencies:
-- python>=3.6
-- pip
-- jinja2
-- click
-- psutil
-- bottleneck
-- netcdf4
-- libnetcdf==4.6.2
-- numpy
-- unidecode
-- dask
-- xarray>=0.12
-- scipy
-- sentry-sdk
-- siphon
-- xclim>=0.12.2
-- pywps>=4.2.3
-
+  - python>=3.6
+  - pip
+  - jinja2
+  - click
+  - psutil
+  - bottleneck
+  - netcdf4
+  - libnetcdf==4.6.2
+  - numpy
+  - unidecode
+  - dask
+  - xarray>=0.12
+  - scipy
+  - sentry-sdk
+  - siphon
+  - xclim>=0.12.2
+  - pywps~=4.2.3

finch/processes/__init__.py

@@ -1,6 +1,7 @@
 from .wps_xsubsetbbox import SubsetBboxProcess
 from .wps_xsubsetpoint import SubsetGridPointProcess
-from .wps_xsubset_bccaqv2 import SubsetBCCAQV2Process
+from .wps_xsubsetpoint_bccaqv2 import SubsetGridPointBCCAQV2Process
+from .wps_xsubsetbbox_bccaqv2 import SubsetBboxBCCAQV2Process
 from .wps_xclim_indices import make_xclim_indicator_process
 from .wps_bccaqv2_heatwave import BCCAQV2HeatWave
 import xclim
@@ -29,7 +30,8 @@ def get_indicators(*args):
     [
         SubsetBboxProcess(),
         SubsetGridPointProcess(),
-        SubsetBCCAQV2Process(),
+        SubsetGridPointBCCAQV2Process(),
+        SubsetBboxBCCAQV2Process(),
         BCCAQV2HeatWave(),
     ]
 )

finch/processes/subset.py

@@ -44,6 +44,8 @@ def process_resource(resource):
         if threads > 1:
             pool = ThreadPool(processes=threads)
             list(pool.imap_unordered(process_resource, resources))
+            pool.close()
+            pool.join()
         else:
             for r in resources:
                 process_resource(r)

finch/processes/utils.py

@@ -7,6 +7,7 @@
 from typing import List, Tuple
 from enum import Enum
 
+import numpy as np
 import pandas as pd
 import xarray as xr
 import requests
@@ -241,9 +242,14 @@ def get_attrs_fallback(ds, *args):
 
             ds = ds.rename({variable: output_variable})
 
-            df = ds.to_dataframe()[["lat", "lon", output_variable]]
-            # most runs have timestamp with hour == 12 a few hour == 0 .. make uniform
-            df.index = df.index.map(lambda x: x.replace(hour=12))
+            # most runs have timestamp with hour == 12 a few hour == 0 ... make uniform
+            if not np.all(ds.time.dt.hour == 12):
+                attrs = ds.time.attrs
+                ds['time'] = [y.replace(hour=12) for y in ds.time.values]
+                ds.time.attrs = attrs
+
+            df = ds.to_dataframe()
+            df = df.reset_index().set_index('time')[["lat", "lon", output_variable]]
 
             if calendar not in concat_by_calendar:
                 concat_by_calendar[calendar] = [df]
@@ -255,7 +261,8 @@ def get_attrs_fallback(ds, *args):
     output_csv_list = []
     for calendar_type, data in concat_by_calendar.items():
         output_csv = output_folder / f"{filename_prefix}_{calendar_type}.csv"
-        pd.concat(data, axis=1).to_csv(output_csv)
+        dropna_threshold = 3  # lat + lon + at least one value
+        pd.concat(data, axis=1).dropna(axis=1, thresh=dropna_threshold).to_csv(output_csv)
         output_csv_list.append(output_csv)
 
     metadata_folder = output_folder / "metadata"

finch/processes/wps_xsubsetbbox.py

@@ -111,8 +111,6 @@ def subset(
         lat0 = wps_inputs["lat0"][0].data
         lon1 = self.get_input_or_none(wps_inputs, "lon1")
         lat1 = self.get_input_or_none(wps_inputs, "lat1")
-        # dt0 = wps_inputs['dt0'][0].data or None
-        # dt1 = wps_inputs['dt1'][0].data or None
         start = self.get_input_or_none(wps_inputs, "start_date")
         end = self.get_input_or_none(wps_inputs, "end_date")
         variables = [r.data for r in wps_inputs.get("variable", [])]

finch/processes/wps_xsubset_bccaqv2.py → finch/processes/wps_xsubsetbbox_bccaqv2.py

@@ -2,22 +2,26 @@
 from pywps.response.execute import ExecuteResponse
 from pywps.app.exceptions import ProcessError
 from pywps.app import WPSRequest
+from .wpsio import start_date, end_date
 from pywps import LiteralInput, ComplexOutput, FORMATS, configuration
 
 from finch.processes import SubsetBboxProcess
 from finch.processes.subset import SubsetProcess
 from finch.processes.utils import get_bccaqv2_inputs, netcdf_to_csv, zip_files
 
 
-class SubsetBCCAQV2Process(SubsetBboxProcess):
+class SubsetBboxBCCAQV2Process(SubsetBboxProcess):
     """Subset a NetCDF file using bounding box geometry."""
 
     def __init__(self):
         inputs = [
             LiteralInput(
                 "variable",
                 "NetCDF Variable",
-                abstract="Name of the variable in the NetCDF file.",
+                abstract=(
+                    "Name of the variable in the NetCDF file."
+                    "If not provided, all variables will be subsetted."
+                ),
                 data_type="string",
                 default=None,
                 min_occurs=0,
@@ -49,49 +53,21 @@ def __init__(self):
             LiteralInput(
                 "lat1",
                 "Maximum latitude",
-                abstract="Maximum latitude. Omit this coordinate to subset for a single grid cell.",
+                abstract="Maximum latitude",
                 data_type="float",
                 default=None,
                 min_occurs=0,
             ),
             LiteralInput(
                 "lon1",
                 "Maximum longitude",
-                abstract="Maximum longitude. Omit this coordinate to subset for a single grid cell.",
+                abstract="Maximum longitude",
                 data_type="float",
                 default=None,
                 min_occurs=0,
             ),
-            # LiteralInput('dt0',
-            #              'Initial datetime',
-            #              abstract='Initial datetime for temporal subsetting. Defaults to first date in file.',
-            #              data_type='dateTime',
-            #              default=None,
-            #              min_occurs=0,
-            #              max_occurs=1),
-            # LiteralInput('dt1',
-            #              'Final datetime',
-            #              abstract='Final datetime for temporal subsetting. Defaults to last date in file.',
-            #              data_type='dateTime',
-            #              default=None,
-            #              min_occurs=0,
-            #              max_occurs=1),
-            LiteralInput(
-                "y0",
-                "Initial year",
-                abstract="Initial year for temporal subsetting. Defaults to first year in file.",
-                data_type="integer",
-                default=None,
-                min_occurs=0,
-            ),
-            LiteralInput(
-                "y1",
-                "Final year",
-                abstract="Final year for temporal subsetting. Defaults to last year in file.",
-                data_type="integer",
-                default=None,
-                min_occurs=0,
-            ),
+            start_date,
+            end_date,
             LiteralInput(
                 "output_format",
                 "Output format choice",
@@ -116,8 +92,8 @@ def __init__(self):
         SubsetProcess.__init__(
             self,
             self._handler,
-            identifier="subset_ensemble_BCCAQv2",
-            title="Subset of BCCAQv2 datasets",
+            identifier="subset_ensemble_bbox_BCCAQv2",
+            title="Subset of BCCAQv2 datasets, using a bounding box",
             version="0.1",
             abstract=(
                 "For the BCCAQv2 datasets, "
@@ -134,13 +110,13 @@ def __init__(self):
     def _handler(self, request: WPSRequest, response: ExecuteResponse):
         self.write_log("Processing started", response, 5)
 
+        # Build output filename
         variable = self.get_input_or_none(request.inputs, "variable")
         rcp = self.get_input_or_none(request.inputs, "rcp")
         lat0 = self.get_input_or_none(request.inputs, "lat0")
         lon0 = self.get_input_or_none(request.inputs, "lon0")
         output_format = request.inputs["output_format"][0].data
-
-        output_filename = f"BCCAQv2_subset_{lat0}_{lon0}"
+        output_filename = f"BCCAQv2_subset_bbox_{lat0:.3f}_{lon0:.3f}"
 
         self.write_log("Fetching BCCAQv2 datasets", response, 6)
         request.inputs = get_bccaqv2_inputs(request.inputs, variable, rcp)

finch/processes/wps_xsubsetpoint.py

@@ -1,5 +1,6 @@
 from threading import Lock
 
+import xarray as xr
 from pywps import LiteralInput, ComplexInput, ComplexOutput, FORMATS
 from pywps.inout.outputs import MetaLink4
 from xclim.subset import subset_gridpoint
@@ -8,7 +9,7 @@
 
 
 class SubsetGridPointProcess(SubsetProcess):
-    """Subset a NetCDF file using bounding box geometry."""
+    """Subset a NetCDF file grid cells using a list of coordinates."""
 
     def __init__(self):
         inputs = [
@@ -22,15 +23,15 @@ def __init__(self):
             LiteralInput(
                 "lon",
                 "Longitude",
-                abstract="Longitude coordinate",
-                data_type="float",
+                abstract="Longitude coordinate. Accepts a comma separated list of floats for multiple grid cells.",
+                data_type="string",
                 min_occurs=1,
             ),
             LiteralInput(
                 "lat",
                 "Latitude",
-                abstract="Latitude coordinate.",
-                data_type="float",
+                abstract="Latitude coordinate. Accepts a comma separated list of floats for multiple grid cells.",
+                data_type="string",
                 min_occurs=1,
             ),
             start_date,
@@ -81,10 +82,8 @@ def __init__(self):
         )
 
     def subset(self, wps_inputs, response, start_percentage=10, end_percentage=85, threads=1) -> MetaLink4:
-        lon = wps_inputs["lon"][0].data
-        lat = wps_inputs["lat"][0].data
-        # dt0 = wps_inputs['dt0'][0].data or None
-        # dt1 = wps_inputs['dt1'][0].data or None
+        longitudes = [float(lon) for lon in wps_inputs["lon"][0].data.split(',')]
+        latitudes = [float(lat) for lat in wps_inputs["lat"][0].data.split(',')]
         start = self.get_input_or_none(wps_inputs, "start_date")
         end = self.get_input_or_none(wps_inputs, "end_date")
         variables = [r.data for r in wps_inputs.get("variable", [])]
@@ -108,7 +107,16 @@ def _subset_function(resource):
                 self.write_log(f"Subsetting file {count} of {n_files}", response, percentage)
 
             dataset = dataset[variables] if variables else dataset
-            return subset_gridpoint(dataset, lon=lon, lat=lat, start_date=start, end_date=end)
+
+            global_attributes = dataset.attrs
+            output_ds = None
+            for lon, lat in zip(longitudes, latitudes):
+                subset = subset_gridpoint(dataset, lon=lon, lat=lat, start_date=start, end_date=end)
+                subset = subset.expand_dims(["lat", "lon"])
+                output_ds = output_ds.combine_first(subset) if output_ds is not None else subset
+
+            output_ds.attrs = global_attributes
+            return output_ds
 
         metalink = self.subset_resources(wps_inputs["resource"], _subset_function, threads=threads)