Merge pull request #29 from TUW-GEO/fix_parallel

Fix issues with parallel processing
TUW-GEO · Jun 7, 2024 · 5a27031 · 5a27031
2 parents 308cc89 + fc4b8a3
commit 5a27031
Show file tree

Hide file tree

Showing 3 changed files with 104 additions and 53 deletions.
diff --git a/docs/examples/ts2img.ipynb b/docs/examples/ts2img.ipynb
diff --git a/src/repurpose/ts2img.py b/src/repurpose/ts2img.py
@@ -76,8 +76,7 @@ def _write_img(
  if annual_folder:
  out_path = os.path.join(out_path, f"{dt.year:04}")
 
- if not os.path.exists(out_path):
- os.makedirs(out_path)
+ os.makedirs(out_path, exist_ok=True)
 
  image.attrs['date_created'] = f"File created: {datetime.now()}"
  image.to_netcdf(os.path.join(out_path, filename),
@@ -242,7 +241,7 @@ def _calc_chunk(self, timestamps, preprocess_func=None, preprocess_kwargs=None,
 
  stack = parallel_process_async(
  _convert, ITER_KWARGS, STATIC_KWARGS, n_proc=n_proc,
- show_progress_bars=True, log_path=log_path,
+ show_progress_bars=True, log_path=log_path, backend='threading',
  verbose=False, ignore_errors=self.ignore_errors)
 
  stack = xr.combine_by_coords(stack)

diff --git a/tests/test_ts2img.py b/tests/test_ts2img.py
@@ -102,17 +102,16 @@ def postprocess_func(stack, vars, fillvalue=0):
  variables={'var0': 'var1', 'var2': 'var2'})
 
  with tempfile.TemporaryDirectory() as path_out:
- with pytest.warns(UserWarning): # expected warning about empty stack
- converter.calc(
- path_out, format_out='slice',
- fn_template="test_{datetime}.nc", drop_empty=True,
- preprocess=preprocess_func, preprocess_kwargs={'mult': 2},
- postprocess=postprocess_func, postprocess_kwargs={'vars': ('var2',)},
- encoding={'var1': {'dtype': 'int64', 'scale_factor': 0.0000001,
- '_FillValue': -9999}, },
- var_attrs={'var1': {'long_name': 'test_var1', 'units': 'm'}},
- glob_attrs={'test': 'test'}, var_fillvalues={'var2': -9999},
- var_dtypes={'var2': 'int32'}, n_proc=1)
+ converter.calc(
+ path_out, format_out='slice',
+ fn_template="test_{datetime}.nc", drop_empty=True,
+ preprocess=preprocess_func, preprocess_kwargs={'mult': 2},
+ postprocess=postprocess_func, postprocess_kwargs={'vars': ('var2',)},
+ encoding={'var1': {'dtype': 'int64', 'scale_factor': 0.0000001,
+ '_FillValue': -9999}, },
+ var_attrs={'var1': {'long_name': 'test_var1', 'units': 'm'}},
+ glob_attrs={'test': 'test'}, var_fillvalues={'var2': -9999},
+ var_dtypes={'var2': 'int32'}, n_proc=2)
 
  assert len(os.listdir(os.path.join(path_out, '2020'))) == 28
  assert os.path.isfile(
@@ -208,7 +207,7 @@ def postprocess_func(stack, **kwargs):
  'var2': {'long_name': 'test_var2', 'units': 'm'}},
  glob_attrs={'test': 'test2'},
  var_fillvalues={'var2': -9999},
- var_dtypes={'var2': 'int32'}, n_proc=1)
+ var_dtypes={'var2': 'int32'}, n_proc=2)
 
  # all 10 files must exist, first two emtpy
  assert len(os.listdir(os.path.join(path_out, '2020'))) == 10