Merge pull request #460 from euroargodev/lazy-ftp-open_dataset

Implement laziness with FTP

Author: gmaze

Diff for: argopy/stores/filesystems.py

@@ -19,6 +19,7 @@
 
 log = logging.getLogger("argopy.stores")
 
+
 try:
     from tqdm import tqdm
 except ModuleNotFoundError:
@@ -145,4 +146,4 @@ def new_fs(
     # log.warning(log_msg)
     log.debug(log_msg)
     # log_argopy_callerstack()
-    return fs, cache_registry
+    return fs, cache_registry, fsspec_kwargs

Diff for: argopy/stores/implementations/ftp.py

@@ -29,6 +29,14 @@ class ftpstore(httpstore):
 
     protocol = "ftp"
 
+    @property
+    def host(self):
+        return self.fs.fs.host if self.fs.protocol == 'dir' else self.fs.host
+
+    @property
+    def port(self):
+        return self.fs.fs.port if self.fs.protocol == 'dir' else self.fs.port
+
     def open_dataset(self, url: str,
         errors: Literal["raise", "ignore", "silent"] = "raise",
         lazy: bool = False,
@@ -140,9 +148,7 @@ def load_lazily(
 
             if "ak" not in kwargs:
                 self.ak = ArgoKerchunker()
-                storage_options = {"host": self.fs.fs.host if self.fs.protocol == 'dir' else self.fs.host,
-                                   "port": self.fs.fs.port if self.fs.protocol == 'dir' else self.fs.port}
-                self.ak.storage_options = storage_options
+                self.ak.storage_options = {"host": self.host, "port": self.port}
             else:
                 self.ak = kwargs["ak"]
 
@@ -163,7 +169,7 @@ def load_lazily(
                 return "reference://", xr_opts
             else:
                 warnings.warn(
-                    "This url does not support byte range requests so we cannot load it lazily, falling back on loading in memory."
+                    "This url does not support byte range requests so we cannot load it lazily, falling back on loading in memory.\n(url='%s')" % url
                 )
                 log.debug("This url does not support byte range requests: %s" % self.full_path(url))
                 return load_in_memory(
@@ -187,7 +193,7 @@ def load_lazily(
 
             if "source" not in ds.encoding:
                 if isinstance(url, str):
-                    ds.encoding["source"] = url
+                    ds.encoding["source"] = self.full_path(url)
 
             self.register(url)
             return ds

Diff for: argopy/stores/implementations/gdac.py

@@ -69,7 +69,7 @@ class gdacfs:
 
     @staticmethod
     def path2protocol(path: Union[str, Path]) -> str:
-        """Narrow down any path to a supported protocols, raise GdacPathError if protocol not supported"""
+        """Narrow down any path to a supported protocol, raise GdacPathError if protocol not supported"""
         if isinstance(path, Path):
             return "file"
         else:

Diff for: argopy/stores/implementations/http.py

@@ -300,7 +300,6 @@ def load_lazily(
                 self.ak = ArgoKerchunker()
                 if self.protocol == 's3':
                     storage_options = {'anon': not has_aws_credentials()}
-                    # log.debug(f"AWS credentials: {has_aws_credentials()}")
                 else:
                     storage_options = {}
                 self.ak.storage_options = storage_options
@@ -349,7 +348,7 @@ def load_lazily(
 
             if "source" not in ds.encoding:
                 if isinstance(url, str):
-                    ds.encoding["source"] = url
+                    ds.encoding["source"] = self.full_path(url)
 
             self.register(url)
             return ds

Diff for: argopy/stores/kerchunker.py

@@ -3,6 +3,7 @@
 from pathlib import Path
 import json
 import logging
+from packaging import version
 
 from ..utils import to_list
 from . import memorystore, filestore
@@ -199,19 +200,31 @@ def __repr__(self):
 
     @property
     def store_path(self):
-        """Path to the reference store, including protocol"""
+        """Absolute path to the reference store, including protocol"""
         p = getattr(self.fs, "path", str(Path(".").absolute()))
         # Ensure the protocol is included for non-local files:
-        if self.fs.fs.protocol[0] == "ftp":
-            p = "ftp://" + self.fs.fs.host + fsspec.core.split_protocol(p)[-1]
         if self.fs.fs.protocol[0] == "s3":
             p = "s3://" + fsspec.core.split_protocol(p)[-1]
         return p
 
     def _ncfile2jsfile(self, ncfile):
+        """Convert a netcdf file path to a data store file path for kerchunk zarr reference (json data)"""
         return Path(ncfile).name.replace(".nc", "_kerchunk.json")
 
+    def _ncfile2ncref(self, ncfile: Union[str, Path], fs=None):
+        """Convert a netcdf file path to a key used in internal kerchunk_references"""
+        # return fs.full_path(fs.info(str(ncfile))['name'], protocol=True)
+        return fs.full_path(str(ncfile), protocol=True)
+
     def _magic2chunker(self, ncfile, fs):
+        """Get a netcdf file path chunker alias: 'cdf3' or 'hdf5'
+
+        This is based on the file binary magic value.
+
+        Raises
+        ------
+        :class:`ValueError` if file not recognized
+        """
         magic = fs.open(ncfile).read(3)
         if magic == b"CDF":
             return "cdf3"
@@ -228,6 +241,11 @@ def nc2reference(
     ):
         """Compute reference data for a netcdf file (kerchunk json data)
 
+        This method is intended to be used internally, since it's not using the kerchunk reference store.
+
+        Users should rather use the :meth:`to_reference` method to avoid to recompute reference data
+        when available on the :class:`ArgoKerchunker` instance.
+
         Parameters
         ----------
         ncfile : Union[str, Path]
@@ -245,24 +263,27 @@ def nc2reference(
         -------
         dict
         """
-        ncfile_raw = str(ncfile)
         chunker = self._magic2chunker(ncfile, fs) if chunker == "auto" else chunker
+        ncfile_full = self._ncfile2ncref(ncfile, fs=fs)
+
+        storage_options = self.storage_options.copy()
+        if fs.protocol == 'ftp' and version.parse(fsspec.__version__) < version.parse("2024.10.0"):
+            # We need https://github.com/fsspec/filesystem_spec/pull/1673
+            storage_options.pop('host', None)
+            storage_options.pop('port', None)
 
-        ncfile_full = fs.full_path(ncfile_raw, protocol=True)
-        # log.debug(f"Computing kerchunk json zarr references for: {ncfile_full}")
-        # log.debug(f"Kerchunker storage options: {self.storage_options}")
         if chunker == "cdf3":
             chunks = NetCDF3ToZarr(
                 ncfile_full,
                 inline_threshold=self.inline_threshold,
                 max_chunk_size=self.max_chunk_size,
-                storage_options=self.storage_options,
+                storage_options=storage_options,
             )
         elif chunker == "hdf5":
             chunks = SingleHdf5ToZarr(
                 ncfile_full,
                 inline_threshold=self.inline_threshold,
-                storage_options=self.storage_options,
+                storage_options=storage_options,
             )
 
         kerchunk_data = chunks.translate()
@@ -272,7 +293,7 @@ def nc2reference(
         with self.fs.open(kerchunk_jsfile, "wb") as f:
             f.write(json.dumps(kerchunk_data).encode())
 
-        return ncfile, kerchunk_jsfile, kerchunk_data
+        return ncfile_full, kerchunk_jsfile, kerchunk_data
 
     def update_kerchunk_references_from_store(self):
         """Load kerchunk data already on store"""
@@ -362,14 +383,12 @@ def translate(
         return results
 
     def to_reference(self, ncfile: Union[str, Path], fs=None, overwrite: bool = False):
-        """Return zarr reference data for a given netcdf file
-
-        If data are found on the instance file store, load them, otherwise triggers :meth:`ArgoKerchunker.translate` and
-        save data on instance file store.
+        """Return zarr reference data for a given netcdf file path
 
-        This is basically similar to :meth:`ArgoKerchunker.translate` but for a single file and **possibly using pre-computed references**.
+        Return data from the instance store if available, otherwise trigger :meth:`ArgoKerchunker.translate` (which save
+        data on the instance data store).
 
-        This is the method to use in **argopy** file store method :meth:`ArgoStoreProto.open_dataset` to implement laziness.
+        This is the method to use in **argopy** file store methods :meth:`ArgoStoreProto.open_dataset` to implement laziness.
 
         Parameters
         ----------
@@ -388,16 +407,16 @@ def to_reference(self, ncfile: Union[str, Path], fs=None, overwrite: bool = Fals
         """
         if overwrite:
             self.translate(ncfile, fs=fs)
-        elif str(ncfile) not in self.kerchunk_references:
+        elif self._ncfile2ncref(ncfile, fs=fs) not in self.kerchunk_references:
             if self.fs.exists(self._ncfile2jsfile(ncfile)):
                 self.kerchunk_references.update(
-                    {str(ncfile): self._ncfile2jsfile(ncfile)}
+                    {self._ncfile2ncref(ncfile, fs=fs): self._ncfile2jsfile(ncfile)}
                 )
             else:
                 self.translate(ncfile, fs=fs)
 
         # Read and load the kerchunk JSON file:
-        kerchunk_jsfile = self.kerchunk_references[str(ncfile)]
+        kerchunk_jsfile = self.kerchunk_references[self._ncfile2ncref(ncfile, fs=fs)]
         with self.fs.open(kerchunk_jsfile, "r") as file:
             kerchunk_data = json.load(file)
 
@@ -406,18 +425,18 @@ def to_reference(self, ncfile: Union[str, Path], fs=None, overwrite: bool = Fals
             target_ok = False
             for key, value in kerchunk_data["refs"].items():
                 if key not in [".zgroup", ".zattrs"] and "0." in key:
-                    if value[0] == ncfile:
+                    if value[0] == self._ncfile2ncref(ncfile, fs=fs):
                         target_ok = True
                         break
             if not target_ok:
                 kerchunk_data = self.to_reference(ncfile, overwrite=True, fs=fs)
 
         return kerchunk_data
 
-    def pprint(self, ncfile: Union[str, Path], params: List[str] = None):
+    def pprint(self, ncfile: Union[str, Path], params: List[str] = None, fs=None):
         """Pretty print kerchunk json data for a netcdf file"""
         params = to_list(params) if params is not None else []
-        kerchunk_data = self.to_reference(ncfile)
+        kerchunk_data = self.to_reference(ncfile, fs=fs)
 
         # Pretty print JSON data
         keys_to_select = [".zgroup", ".zattrs", ".zmetadata"]
@@ -463,4 +482,4 @@ def supported(self, ncfile: Union[str, Path], fs=None) -> bool:
             return fs.first(ncfile) is not None
         except Exception:
             log.debug(f"Could not read {ncfile} with {fs}")
-            return False
+            return False

Diff for: argopy/stores/spec.py

@@ -7,9 +7,10 @@
 import pickle  # nosec B403 only used with internal files/assets
 import json
 import tempfile
+import aiohttp
 from typing import Union
 from pathlib import Path
-import aiohttp
+import logging
 
 
 from ..options import OPTIONS
@@ -20,6 +21,9 @@
 from .filesystems import new_fs
 
 
+log = logging.getLogger("argopy.stores.spec")
+
+
 class ArgoStoreProto(ABC):
     """Argo File System Prototype
 
@@ -45,7 +49,7 @@ def __init__(self, cache: bool = False, cachedir: str = "", **kwargs):
         self.cache = cache
         self.cachedir = OPTIONS["cachedir"] if cachedir == "" else cachedir
         self._fsspec_kwargs = {**kwargs}
-        self.fs, self.cache_registry = new_fs(
+        self.fs, self.cache_registry, self._fsspec_kwargs = new_fs(
             self.protocol, self.cache, self.cachedir, **self._fsspec_kwargs
         )
 
@@ -115,9 +119,19 @@ def is_read(uri):
         else:
             return None
 
-    def expand_path(self, path):
+    def expand_path(self, path, **kwargs):
+        """Turn one or more globs or directories into a list of all matching paths to files or directories.
+
+        For http store, return path unchanged (not implemented).
+
+        kwargs are passed to fsspec expand_path which call ``glob`` or ``find``, which may in turn call ``ls``.
+
+        Returns
+        -------
+        list
+        """
         if self.protocol != "http" and self.protocol != "https":
-            return self.fs.expand_path(path)
+            return self.fs.expand_path(path, **kwargs)
         else:
             return [path]
 
@@ -139,6 +153,8 @@ def full_path(self, path, protocol: bool = False):
 
         """
         fp = getattr(self.fs, '_join', lambda x: x)(path)
+        if self.protocol == 'ftp':
+            fp = f"{self.host}:{self.port}{self.fs._strip_protocol(fp)}"
         if not protocol:
             return fp
         else:

Diff for: argopy/tests/test_stores_fsspec.py

@@ -55,12 +55,12 @@
 class Test_new_fs:
 
     def test_default(self):
-        fs, cache_registry = new_fs()
+        fs, cache_registry, fsspec_kwargs = new_fs()
         assert id_implementation(fs) is not None
         assert is_initialised(cache_registry)
 
     def test_cache_type(self):
-        fs, cache_registry = new_fs(cache=True)
+        fs, cache_registry, fsspec_kwargs = new_fs(cache=True)
         assert id_implementation(fs) == ['filecache']
 
 

Diff for: docs/whats-new.rst

@@ -7,6 +7,19 @@ What's New
 
 |pypi dwn| |conda dwn|
 
+Coming up next
+--------------
+
+Internals
+^^^^^^^^^
+
+- **Open netcdf files lazily from ftp server**. Adding to s3 and http, we now support laziness with ftp using `kerchunk <https://fsspec.github.io/kerchunk/>`_. Checkout the dedicated :ref:`lazy` section of the documentation. (:pr:`460`) by |gmaze|.
+
+Energy
+^^^^^^
+
+|eqco2_since_last_release|
+
 
 v1.1.0 (18 March 2025)
 ----------------------
@@ -103,7 +116,7 @@ Internals
 Energy
 ^^^^^^
 
-|eqco2_since_last_release|
+.. image:: https://img.shields.io/badge/Total%20carbon%20emitted%20by%20release%20v1.1.0%20%5BgCO2eq%5D-1328.99-black?style=plastic&labelColor=grey
 
 
 v1.0.0 (16 Oct. 2024)