First version of dataset verification, addressing #17

Author: forman

nc2zarr/cli.py

@@ -53,6 +53,10 @@
 @click.option('--sort-by', '-s', 'sort_by', default=None,
               type=click.Choice(['path', 'name'], case_sensitive=True),
               help='Sort input files by specified property.')
+@click.option('--verify', 'verify', type=click.Choice(["on", 'off', 'auto']),
+              default='auto',
+              help='Switch verification either on, or off,'
+                   ' or leave it up to CONFIG_FILE (=auto, the default).')
 @click.option('--dry-run', '-d', 'dry_run', is_flag=True, default=None,
               help='Open and process inputs only, omit data writing.')
 @click.option('--verbose', '-v', 'verbose', is_flag=True, multiple=True,
@@ -68,6 +72,7 @@ def nc2zarr(input_paths: Tuple[str],
             append: bool,
             decode_cf: bool,
             sort_by: str,
+            verify: str,
             dry_run: bool,
             verbose: Tuple[bool],
             version: bool):
@@ -129,6 +134,7 @@ def nc2zarr(input_paths: Tuple[str],
                                     output_path=output_path,
                                     output_overwrite=overwrite,
                                     output_append=append,
+                                    verify_enabled=True if verify == 'on' else False if verify == 'off' else None,
                                     verbosity=sum(verbose) if verbose else None,
                                     dry_run=dry_run)
         Converter(**config_kwargs).run()

nc2zarr/converter.py

@@ -31,6 +31,7 @@
 from .preprocessor import DatasetPreProcessor
 from .processor import DatasetProcessor
 from .writer import DatasetWriter
+from .verifier import DatasetVerifier
 
 
 class Converter:
@@ -55,6 +56,8 @@ class Converter:
     :param output_append:
     :param output_append_dim:
     :param output_s3:
+    :param verify_enabled:
+    :param verify_open_params:
     :param dry_run:
     :param verbosity:
     """
@@ -79,6 +82,8 @@ def __init__(self,
                  output_append_dim: str = None,
                  output_s3: Dict[str, Any] = None,
                  output_retry: Dict[str, Any] = None,
+                 verify_enabled: bool = None,
+                 verify_open_params: Dict[str, Any] = None,
                  dry_run: bool = False,
                  verbosity: int = None):
 
@@ -114,6 +119,8 @@ def __init__(self,
         self.output_append_dim = output_append_dim
         self.output_s3 = output_s3
         self.output_retry = output_retry
+        self.verify_enabled = verify_enabled
+        self.verify_open_params = verify_open_params
         self.dry_run = dry_run
         self.verbosity = verbosity
 
@@ -153,9 +160,16 @@ def _run(self):
                                dry_run=self.dry_run,
                                reset_attrs=not self.input_decode_cf)
 
+        dataset_verifier = DatasetVerifier(output_path=self.output_path,
+                                           output_s3_kwargs=self.output_s3,
+                                           verify_enabled=self.verify_enabled,
+                                           verify_open_params=self.verify_open_params)
+
         append = None
         for input_dataset in opener.open_datasets(preprocess=pre_processor.preprocess_dataset):
             output_dataset, output_encoding = processor.process_dataset(input_dataset)
             writer.write_dataset(output_dataset, encoding=output_encoding, append=append)
             input_dataset.close()
             append = True
+
+        dataset_verifier.verify_dataset()

nc2zarr/res/config-template.yml

@@ -163,3 +163,11 @@ output:
     delay: 0.1
     # multiplier applied to delay between attempts. 1.0 means no backoff.
     backoff: 1.1
+
+
+verify:
+  enabled: true
+
+  open_params:
+    consolidated: false
+    decode_cf: false

nc2zarr/verifier.py

@@ -0,0 +1,68 @@
+# The MIT License (MIT)
+# Copyright (c) 2021 by Brockmann Consult GmbH and contributors
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy of
+# this software and associated documentation files (the "Software"), to deal in
+# the Software without restriction, including without limitation the rights to
+# use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
+# of the Software, and to permit persons to whom the Software is furnished to do
+# so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+from typing import Dict, Any
+
+import fsspec
+import fsspec.implementations.local
+import xarray as xr
+
+from .log import LOGGER
+from .log import log_duration
+
+
+class DatasetVerifier:
+    def __init__(self,
+                 output_path: str,
+                 output_s3_kwargs: Dict[str, Any] = None,
+                 verify_enabled: Dict[str, Any] = None,
+                 verify_open_params: Dict[str, Any] = None,
+                 dry_run: bool = False):
+        if not output_path:
+            raise ValueError('output_path must be given')
+        self._output_path = output_path
+        self._output_s3_kwargs = output_s3_kwargs
+        self._verify_enabled = verify_enabled
+        self._verify_open_params = verify_open_params or {}
+        self._dry_run = dry_run
+
+    def verify_dataset(self):
+
+        if not self._verify_enabled:
+            LOGGER.info('Dataset verification disabled.')
+            return
+
+        with log_duration(f'Verifying dataset'):
+            if not self._dry_run:
+                if self._output_s3_kwargs or self._output_path.startswith('s3://'):
+                    fs = fsspec.filesystem('s3', **(self._output_s3_kwargs or {}))
+                else:
+                    fs = fsspec.filesystem('file')
+                store = fs.get_mapper(self._output_path, check=False, create=False)
+                # noinspection PyBroadException
+                try:
+                    dataset = xr.open_zarr(store, **self._verify_open_params)
+                    LOGGER.info(dataset)
+                    LOGGER.info('Dataset verification passed.')
+                except BaseException as e:
+                    LOGGER.error('Dataset verification failed!')
+            else:
+                LOGGER.info('Dataset verification skipped, it is a dry run.')