Add a GRIB backend via ECMWF cfgrib / ecCodes

xarray/backends/api.py

@@ -297,7 +297,7 @@ def maybe_decode_store(store, lock=False):
             store = backends.PseudoNetCDFDataStore.open(
                 filename_or_obj, lock=lock, **backend_kwargs)
         elif engine == 'cfgrib':
-            store = backends.CfGribDataStore.from_path(
+            store = backends.CfGribDataStore(
                 filename_or_obj, lock=lock, **backend_kwargs)
         else:
             raise ValueError('unrecognized engine for open_dataset: %r'

xarray/backends/cfgrib_.py

@@ -25,6 +25,12 @@
 from ..core import indexing
 from ..core.utils import Frozen, FrozenOrderedDict
 from .common import AbstractDataStore, BackendArray
+from .file_manager import CachingFileManager
+from .locks import ensure_lock, SerializableLock
+
+# FIXME: Add a dedicated lock just in case, even if ecCodes is supposed to be thread-safe in most
+# circumstances. See: https://confluence.ecmwf.int/display/ECC/Frequently+Asked+Questions
+ECCODES_LOCK = SerializableLock()
 
 
 class CfGribArrayWrapper(BackendArray):
@@ -50,28 +56,29 @@ class CfGribDataStore(AbstractDataStore):
     """
     Implements the ``xr.AbstractDataStore`` read-only API for a GRIB file.
     """
-    def __init__(self, ds, lock=False):
-        self.ds = ds
-        self.lock = lock
-
-    @classmethod
-    def from_path(cls, path, lock=False, **backend_kwargs):
+    def __init__(self, filename, lock=None, **backend_kwargs):
         import cfgrib
-        return cls(ds=cfgrib.open_file(path, **backend_kwargs), lock=lock)
+        if lock is None:
+            lock = ECCODES_LOCK
+        self.lock = ensure_lock(lock)
+        backend_kwargs['filter_by_keys'] = tuple(backend_kwargs.get('filter_by_keys', {}).items())
+        self._manager = CachingFileManager(
+            cfgrib.open_file, filename, lock=lock, mode='r', kwargs=backend_kwargs)
+
+    @property
+    def ds(self):
+        return self._manager.acquire()
 
     def open_store_variable(self, name, var):
         if isinstance(var.data, np.ndarray):
             data = var.data
         else:
             data = indexing.LazilyOuterIndexedArray(CfGribArrayWrapper(var.data))
 
-        dimensions = var.dimensions
-        attrs = var.attributes
-
         encoding = self.ds.encoding.copy()
         encoding['original_shape'] = var.data.shape
 
-        return Variable(dimensions, data, attrs, encoding)
+        return Variable(var.dimensions, data, var.attributes, encoding)
 
     def get_variables(self):
         return FrozenOrderedDict((k, self.open_store_variable(k, v))
@@ -87,3 +94,6 @@ def get_encoding(self):
         encoding = {}
         encoding['unlimited_dims'] = {k for k, v in self.ds.dimensions.items() if v is None}
         return encoding
+
+    def close(self):
+        self._manager.close()