Fix conflicts with master

.github/workflows/build.yml

@@ -84,9 +84,9 @@ jobs:
           fi
           ls .artifacts/dist
       - name: Upload Artifacts
-        uses: actions/upload-artifact@v2
+        uses: actions/upload-artifact@v4
         with:
-          name: Artifacts
+          name: Artifacts-${{ matrix.python-version }}-${{ matrix.os }}
           path: .artifacts/*
   coveralls:
     name: Submit Coveralls 👚
@@ -111,6 +111,10 @@ jobs:
           echo "GITHUB_REPOSITORY = $GITHUB_REPOSITORY"
       - name: Download Artifacts
         uses: actions/download-artifact@v4
+        with:
+          path: Artifacts
+          pattern: Artifacts-*
+          merge-multiple: true
       - name: Display downloaded files
         run: ls -aR
       - name: Upload to PyPI

CHANGELOG.rst

@@ -4,7 +4,12 @@ Changelog
 
 Unreleased changes in master branch
 ===================================
+-
 
+Version 0.11
+============
+
+- Use joblib for parallel processing framework, improved logging
 - Added option to parallelize Img2Ts process
 - Fix bug where a wrong grid attribute was used.
 

README.rst

@@ -2,23 +2,25 @@
 repurpose
 =========
 
-.. image:: https://github.com/TUW-GEO/repurpose/actions/workflows/build.yml/badge.svg?branch=master
+|ci| |cov| |pip| |doc|
+
+.. |ci| image:: https://github.com/TUW-GEO/repurpose/actions/workflows/build.yml/badge.svg?branch=master
    :target: https://github.com/TUW-GEO/repurpose/actions
 
-.. image:: https://coveralls.io/repos/github/TUW-GEO/repurpose/badge.svg?branch=master
+.. |cov| image:: https://coveralls.io/repos/github/TUW-GEO/repurpose/badge.svg?branch=master
    :target: https://coveralls.io/github/TUW-GEO/repurpose?branch=master
 
-.. image:: https://badge.fury.io/py/repurpose.svg
+.. |pip| image:: https://badge.fury.io/py/repurpose.svg
     :target: http://badge.fury.io/py/repurpose
 
-.. image:: https://readthedocs.org/projects/repurpose/badge/?version=latest
+.. |doc| image:: https://readthedocs.org/projects/repurpose/badge/?version=latest
    :target: http://repurpose.readthedocs.org/
 
 
 This package provides routines for the conversion of image formats to time
-series and vice versa. It is part of the `poets project
-<http://tuw-geo.github.io/poets/>`_ and works best with the readers and writers
-supported there. The main use case is for data that is sampled irregularly in
+series and vice versa. It works best with the readers and writers
+supported by `pynetcf <https://github.com/TUW-GEO/pynetcf>`_.
+The main use case is for data that is sampled irregularly in
 space or time. If you have data that is sampled in regular intervals then there
 are alternatives to this package which might be better for your use case. See
 `Alternatives`_ for more detail.
@@ -67,8 +69,9 @@ It includes two main modules:
 - ``img2ts`` for image/swath to time series conversion, including support for
   spatial resampling.
 - ``ts2img`` for time series to image conversion, including support for temporal
-  resampling. This module is very experimental at the moment.
+  resampling.
 - ``resample`` for spatial resampling of (regular or irregular) gridded data to different resolutions.
+- ``process`` contains a framework for parallel processing, error handling and logging based on `joblib <https://github.com/joblib/joblib>`_
 
 Alternatives
 ============
@@ -97,7 +100,8 @@ against our master branch for new features or bug fixes.
 Development setup
 -----------------
 
-For Development we recommend a ``conda`` environment
+For Development we recommend a ``conda`` environment with all packages listed
+in `environment.yml`.
 
 Guidelines
 ----------

setup.cfg

@@ -6,12 +6,12 @@
 name = repurpose
 description = Package for image to timeseries to image conversion
 author = TU Wien
-author-email = [email protected]
+author_email = [email protected]
 license = mit
-long-description = file: README.rst
-long-description-content-type = text/x-rst; charset=UTF-8
+long_description = file: README.rst
+long_description_content_type = text/x-rst; charset=UTF-8
 url = https://repurpose.readthedocs.io/en/latest/
-project-urls =
+project_urls =
     Documentation = https://repurpose.readthedocs.io/en/latest/
 # Change if running only on Windows, Mac or Linux (comma-separated)
 platforms = any

src/repurpose/img2ts.py

@@ -1,8 +1,6 @@
 from repurpose.process import parallel_process_async, idx_chunks
-from typing import List
 import pynetcf.time_series as nc
-from pygeogrids.grids import CellGrid, BasicGrid
-import pygeogrids.grids as grids
+from pygeogrids.grids import CellGrid
 import repurpose.resample as resamp
 import numpy as np
 import os
@@ -577,6 +575,8 @@ def calc(self):
 
         for img_stack_dict, timestamps in self.img_bulk():
             # =================================================================
+            logging.info(f"Finished reading bulk with {len(timestamps)} images")
+
             start_time = datetime.now()
 
             # temporally drop grids, due to issue when pickling them...
@@ -597,6 +597,7 @@ def calc(self):
 
             keys = list(img_stack_dict.keys())
             for key in keys:
+                #print(key)
                 # rename variable in output dataset
                 if self.variable_rename is None:
                     var_new_name = str(key)

src/repurpose/ts2img.py

@@ -42,8 +42,15 @@ def _convert(converter: 'Ts2Img',
         if ts is None:
             continue
         if preprocess_func is not None:
-            preprocess_kwargs = preprocess_kwargs or {}
-            ts = preprocess_func(ts, **preprocess_kwargs)
+            preprocess_func = np.atleast_1d(preprocess_func)
+            if preprocess_kwargs is None:
+                preprocess_kwargs = [{}] * len(preprocess_func)
+            preprocess_kwargs = np.atleast_1d(preprocess_kwargs)
+            if len(preprocess_func) != len(preprocess_kwargs):
+                raise ValueError("Length of preprocess_func and "
+                                 "preprocess_kwargs is different")
+            for func, kwargs in zip(preprocess_func, preprocess_kwargs):
+                ts = func(ts, **kwargs)
         if np.any(np.isin(ts.columns, Ts2Img._protected_vars)):
             raise ValueError(
                 f"Time series contains protected variables. "
@@ -269,7 +276,7 @@ def calc(self, path_out, format_out='slice', preprocess=None,
             - stack: write all time steps into one file. In this case if there
                 is a {datetime} placeholder in the fn_template, then the time
                 range is inserted.
-        preprocess: callable, optional (default: None)
+        preprocess: callable or list[Callable], optional (default: None)
             Function that is applied to each time series before converting it.
             The first argument is the data frame that the reader returns.
             Additional keyword arguments can be passed via `preprocess_kwargs`.
@@ -285,25 +292,25 @@ def preprocess_add(df: pd.DataFrame, **preprocess_kwargs) \
                 df['var3'] = df['var1'] + df['var2']
                 return df
             ```
-        preprocess_kwargs: dict, optional (default: None)
+        preprocess_kwargs: dict or list[dict], optional (default: None)
             Keyword arguments for the preprocess function. If None are given,
             then the preprocessing function is is called with only the input
             data frame and no additional arguments (see example above).
-        postprocess: Callable, optional (default: None)
-            Function that is applied to the image stack after loading the data
-            and before writing it to disk. The function must take xarray
+        postprocess: Callable or list[Callable], optional (default: None)
+            Function(s) applied to the image stack after loading the data
+            and before writing it to disk. The function must take an xarray
             Dataset as the first argument and return an xarray Dataset of the
-            same form as the input data.
+            same form.
             A simple example for a preprocessing function to add a new variable
             from the sum of two existing variables:
             ```
-            def preprocess_add(stack: xr.Dataset, **postprocess_kwargs) \
+            def postprocess_add(stack: xr.Dataset, **postprocess_kwargs) \
                     -> xr.Dataset
                 stack = stack.assign(var3=lambda x: x['var0'] + x['var2'])
                 return stack
             ```
-        postprocess_kwargs: dict, optional (default: None)
-            Keyword arguments for the postprocess function. If None are given,
+        postprocess_kwargs: dict or list[dict], optional (default: None)
+            Keyword arguments for the postprocess function(s). If None are given,
             then the postprocess function is called with only the input
             image stack and no additional arguments (see example above).
         fn_template: str, optional (default: "{datetime}.nc")
@@ -387,10 +394,6 @@ def preprocess_add(stack: xr.Dataset, **postprocess_kwargs) \
 
                 self.stack = self.stack.drop_isel(time=idx_empty)
 
-            if postprocess is not None:
-                postprocess_kwargs = postprocess_kwargs or {}
-                self.stack = postprocess(self.stack, **postprocess_kwargs)
-
             if var_fillvalues is not None:
                 for var, fillvalue in var_fillvalues.items():
                     self.stack[var].values = np.nan_to_num(
@@ -416,6 +419,19 @@ def preprocess_add(stack: xr.Dataset, **postprocess_kwargs) \
                 for var in var_attrs:
                     self.stack[var].attrs.update(var_attrs[var])
 
+            if postprocess is not None:
+                # this is done as late as possible
+                postprocess = np.atleast_1d(postprocess)
+                if postprocess_kwargs is None:
+                    postprocess_kwargs = [{}] * len(postprocess)
+                postprocess_kwargs = np.atleast_1d(postprocess_kwargs)
+                if len(postprocess) != len(postprocess_kwargs):
+                    raise ValueError("postprocess and postprocess_kwargs "
+                                     "have different lengths")
+
+                for func, kwargs in zip(postprocess, postprocess_kwargs):
+                    self.stack = func(self.stack, **kwargs)
+
             if self.stack['time'].size == 0:
                 warnings.warn("No images in stack to write to disk.")
                 self.stack = None