Support for zarr

[gmaze]

In this PR we shall provide support for exporting and reading Argo zarr archives to/from local or remote netcdf datasets

A simple netcdf to zarr converter would be:

from argopy import DataFetcher
ds = DataFetcher(src='gdac', mode='expert').float(6903091).to_xarray()
ds.argo.to_zarr("6903091_prof.zarr")

Data variable types and compression would be automatically handled by argopy.

This converter shall allows for a simple syntax like this:

import xarray as xr
ds = xr.open_zarr("6903091_prof.zarr")  # No need for engine='argo' to cast variables appropriately

Note
This PR is one step toward support data fetching from a zarr GDAC on s3, assuming the GDAC data structure is not modified:

./dac/<DAC>/<WMO>/*.nc
./dac/<DAC>/<WMO>/profiles/*.nc

to:

./dac/<DAC>/<WMO>/*.zarr
./dac/<DAC>/<WMO>/profiles/*.zarr

So that a file path such as
/pub/dac/coriolis/6903091/6903091_prof.nc
goes to
/pub/etc/ArgoZarr/dac/coriolis/6903091/6903091_prof.zarr

This is alive, but experimental on S3 prototype.

You can browse it at:

https://argo-gdac-sandbox.s3.eu-west-3.amazonaws.com/pub/index.html#pub/etc/ArgoZarr/

[codecov]

❌ 17 Tests Failed:

Tests completed	Failed	Passed	Skipped
1680	17	1663	546

View the top 3 failed tests by shortest run time

test_fetchers_data_gdac.py::TestBackend::test_fetching_cached[host='s3', ds='phy', mode='expert', {'float': [13857]}]

Stack Traces | 0.274s run time

self = <argopy.tests.test_fetchers_data_gdac.TestBackend object at 0x000001DA305071D0>
mocked_httpserver = 'http://127.0.0.1:9898'
cached_fetcher = <datafetcher.gdac>
#x1F310 Name: Ifremer GDAC Argo data fetcher for floats
#x1F916 Domain: WMO13857
#x1F517 API: s3:.../argo-gdac-sandbox/...searched: True (98 matches, 100.0000%)
#x1F3C4 User mode: expert
#x1F7E1+#x1F535 Dataset: phy
#x1F324  Performances: cache=True, parallel=False

    @pytest.mark.parametrize("cached_fetcher", VALID_ACCESS_POINTS, indirect=True, ids=VALID_ACCESS_POINTS_IDS)
    def test_fetching_cached(self, mocked_httpserver, cached_fetcher):
        # Assert the fetcher (this trigger data fetching, hence caching as well):
>       assert_fetcher(mocked_httpserver, cached_fetcher, cacheable=True)

argopy\tests\test_fetchers_data_gdac.py:205: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
argopy\tests\test_fetchers_data_gdac.py:98: in assert_fetcher
    assert isinstance(this_fetcher.to_xarray(errors='raise'), xr.Dataset)
argopy\fetchers.py:618: in to_xarray
    xds = self.fetcher.to_xarray(**kwargs)
argopy\data_fetchers\gdac_data.py:341: in to_xarray
    URI = self.uri  # Call it once
argopy\data_fetchers\gdac_data.py:545: in uri
    self._list_of_argo_files = self.uri_mono2multi(URIs)
argopy\data_fetchers\gdac_data.py:227: in uri_mono2multi
    new_uri = [mono2multi(uri) for uri in URIs]
argopy\data_fetchers\gdac_data.py:227: in <listcomp>
    new_uri = [mono2multi(uri) for uri in URIs]
argopy\data_fetchers\gdac_data.py:200: in mono2multi
    meta = argo_split_path(mono_path)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

this_path = 's3:.../13857/profiles/R13857_001.nc'

    def argo_split_path(this_path):  # noqa C901
        """Split path from a GDAC ftp style Argo netcdf file and return information
    
        >>> argo_split_path('.../6901035/profiles/D6901035_001D.nc')
        >>> argo_split_path('https://data-argo.ifremer..../5903939/profiles/D5903939_103.nc')
    
        Parameters
        ----------
        str
    
        Returns
        -------
        dict
        """
        dacs = [
            "aoml",
            "bodc",
            "coriolis",
            "csio",
            "csiro",
            "incois",
            "jma",
            "kma",
            "kordi",
            "meds",
            "nmdis",
        ]
        output = {}
    
        start_with = (
            lambda f, x: f[0 : len(x)] == x if len(x) <= len(f) else False
        )  # noqa: E731
    
        def split_path(p, sep="/"):
            """Split a pathname.  Returns tuple "(head, tail)" where "tail" is
            everything after the final slash.  Either part may be empty."""
            # Same as posixpath.py but we get to choose the file separator !
            p = os.fspath(p)
            i = p.rfind(sep) + 1
            head, tail = p[:i], p[i:]
            if head and head != sep * len(head):
                head = head.rstrip(sep)
            return head, tail
    
        def fix_localhost(host):
            if "ftp://localhost:" in host:
                return "ftp://%s" % (urlparse(host).netloc)
            if "http://127.0.0.1:" in host:
                return "http://%s" % (urlparse(host).netloc)
            else:
                return ""
    
        known_origins = [
            "https://data-argo.ifremer.fr",
            "ftp://ftp.ifremer.fr/ifremer/argo",
            "ftp://usgodae..../pub/outgoing/argo",
            fix_localhost(this_path),
            "",
        ]
    
        output["origin"] = [
            origin for origin in known_origins if start_with(this_path, origin)
        ][0]
        output["origin"] = "." if output["origin"] == "" else output["origin"] + "/"
        sep = "/" if output["origin"] != "." else os.path.sep
    
        (path, file) = split_path(this_path, sep=sep)
    
        output["path"] = path.replace(output["origin"], "")
        output["name"] = file
    
        # Deal with the path:
        # dac/<DAC>/<FloatWmoID>/
        # dac/<DAC>/<FloatWmoID>/profiles
        path_parts = path.split(sep)
    
        try:
            if path_parts[-1] == "profiles":
                output["type"] = "Mono-cycle profile file"
                output["wmo"] = path_parts[-2]
                output["dac"] = path_parts[-3]
            else:
                output["type"] = "Multi-cycle profile file"
                output["wmo"] = path_parts[-1]
>               output["dac"] = path_parts[-2]
E               IndexError: list index out of range

argopy\utils\format.py:116: IndexError

test_fetchers_data_gdac.py::TestBackend::test_fetching_cached[host='c', ds='phy', mode='expert', {'region': [-20, -16.0, 0, 1, 0, 100.0]}]

Stack Traces | 0.704s run time

self = <argopy.tests.test_fetchers_data_gdac.TestBackend object at 0x000001DA304C2B90>
mocked_httpserver = 'http://127.0.0.1:9898'
cached_fetcher = <datafetcher.gdac>
#x1F310 Name: Ifremer GDAC Argo data fetcher for a space/time region
#x1F5FA  Domain: [x=-20.00/-16.00; y=0.00/...ex searched: True (3 matches, 0.1110%)
#x1F3C4 User mode: expert
#x1F7E1+#x1F535 Dataset: phy
#x1F324  Performances: cache=True, parallel=False

    @pytest.mark.parametrize("cached_fetcher", VALID_ACCESS_POINTS, indirect=True, ids=VALID_ACCESS_POINTS_IDS)
    def test_fetching_cached(self, mocked_httpserver, cached_fetcher):
        # Assert the fetcher (this trigger data fetching, hence caching as well):
>       assert_fetcher(mocked_httpserver, cached_fetcher, cacheable=True)

argopy\tests\test_fetchers_data_gdac.py:205: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
argopy\tests\test_fetchers_data_gdac.py:98: in assert_fetcher
    assert isinstance(this_fetcher.to_xarray(errors='raise'), xr.Dataset)
argopy\fetchers.py:618: in to_xarray
    xds = self.fetcher.to_xarray(**kwargs)
argopy\data_fetchers\gdac_data.py:399: in to_xarray
    results = self.fs.open_mfdataset(URI, **opts)
argopy\stores\filesystems.py:515: in open_mfdataset
    data = future.result()
C:\Users\runneradmin\micromamba\envs\argopy-tests\Lib\concurrent\futures\_base.py:449: in result
    return self.__get_result()
C:\Users\runneradmin\micromamba\envs\argopy-tests\Lib\concurrent\futures\_base.py:401: in __get_result
    raise self._exception
C:\Users\runneradmin\micromamba\envs\argopy-tests\Lib\concurrent\futures\thread.py:58: in run
    result = self.fn(*self.args, **self.kwargs)
argopy\stores\filesystems.py:419: in _mfprocessor
    ds = self.open_dataset(url, **open_dataset_opts)
argopy\stores\filesystems.py:400: in open_dataset
    with self.open(path) as of:
argopy\stores\filesystems.py:232: in open
    return self.fs.open(path, *args, **kwargs)
C:\Users\runneradmin\micromamba\envs\argopy-tests\Lib\site-packages\fsspec\implementations\cached.py:446: in <lambda>
    return lambda *args, **kw: getattr(type(self), item).__get__(self)(
C:\Users\runneradmin\micromamba\envs\argopy-tests\Lib\site-packages\fsspec\spec.py:1303: in open
    f = self._open(
C:\Users\runneradmin\micromamba\envs\argopy-tests\Lib\site-packages\fsspec\implementations\cached.py:446: in <lambda>
    return lambda *args, **kw: getattr(type(self), item).__get__(self)(
C:\Users\runneradmin\micromamba\envs\argopy-tests\Lib\site-packages\fsspec\implementations\cached.py:707: in _open
    self.save_cache()
C:\Users\runneradmin\micromamba\envs\argopy-tests\Lib\site-packages\fsspec\implementations\cached.py:446: in <lambda>
    return lambda *args, **kw: getattr(type(self), item).__get__(self)(
C:\Users\runneradmin\micromamba\envs\argopy-tests\Lib\site-packages\fsspec\implementations\cached.py:206: in save_cache
    self._metadata.save()
C:\Users\runneradmin\micromamba\envs\argopy-tests\Lib\site-packages\fsspec\implementations\cache_metadata.py:227: in save
    self._save(cache, fn)
C:\Users\runneradmin\micromamba\envs\argopy-tests\Lib\site-packages\fsspec\implementations\cache_metadata.py:75: in _save
    with atomic_write(fn, mode="w") as f:
C:\Users\runneradmin\micromamba\envs\argopy-tests\Lib\contextlib.py:144: in __exit__
    next(self.gen)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

path = 'C:\\Users\\RUNNER~1\\AppData\\Local\\Temp\\tmpcf4mglfi\\cache'
mode = 'w'

    @contextlib.contextmanager
    def atomic_write(path: str, mode: str = "wb"):
        """
        A context manager that opens a temporary file next to `path` and, on exit,
        replaces `path` with the temporary file, thereby updating `path`
        atomically.
        """
        fd, fn = tempfile.mkstemp(
            dir=os.path.dirname(path), prefix=os.path.basename(path) + "-"
        )
        try:
            with open(fd, mode) as fp:
                yield fp
        except BaseException:
            with contextlib.suppress(FileNotFoundError):
                os.unlink(fn)
            raise
        else:
>           os.replace(fn, path)
E           PermissionError: [WinError 5] Access is denied: 'C:\\Users\\RUNNER~1\\AppData\\Local\\Temp\\tmpcf4mglfi\\cache-sa515b1_' -> 'C:\\Users\\RUNNER~1\\AppData\\Local\\Temp\\tmpcf4mglfi\\cache'

C:\Users\runneradmin\micromamba\envs\argopy-tests\Lib\site-packages\fsspec\utils.py:625: PermissionError

test_fetchers_data_gdac.py::TestBackend::test_fetching_cached[host='c', ds='phy', mode='research', {'region': [-20, -16.0, 0, 1, 0, 100.0]}]

Stack Traces | 0.709s run time

self = <argopy.tests.test_fetchers_data_gdac.TestBackend object at 0x000001DA304C27D0>
mocked_httpserver = 'http://127.0.0.1:9898'
cached_fetcher = <datafetcher.gdac>
#x1F310 Name: Ifremer GDAC Argo data fetcher for a space/time region
#x1F5FA  Domain: [x=-20.00/-16.00; y=0.00/... searched: True (3 matches, 0.1110%)
#x1F6A3 User mode: research
#x1F7E1+#x1F535 Dataset: phy
#x1F324  Performances: cache=True, parallel=False

    @pytest.mark.parametrize("cached_fetcher", VALID_ACCESS_POINTS, indirect=True, ids=VALID_ACCESS_POINTS_IDS)
    def test_fetching_cached(self, mocked_httpserver, cached_fetcher):
        # Assert the fetcher (this trigger data fetching, hence caching as well):
>       assert_fetcher(mocked_httpserver, cached_fetcher, cacheable=True)

argopy\tests\test_fetchers_data_gdac.py:205: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
argopy\tests\test_fetchers_data_gdac.py:98: in assert_fetcher
    assert isinstance(this_fetcher.to_xarray(errors='raise'), xr.Dataset)
argopy\fetchers.py:618: in to_xarray
    xds = self.fetcher.to_xarray(**kwargs)
argopy\data_fetchers\gdac_data.py:399: in to_xarray
    results = self.fs.open_mfdataset(URI, **opts)
argopy\stores\filesystems.py:515: in open_mfdataset
    data = future.result()
C:\Users\runneradmin\micromamba\envs\argopy-tests\Lib\concurrent\futures\_base.py:449: in result
    return self.__get_result()
C:\Users\runneradmin\micromamba\envs\argopy-tests\Lib\concurrent\futures\_base.py:401: in __get_result
    raise self._exception
C:\Users\runneradmin\micromamba\envs\argopy-tests\Lib\concurrent\futures\thread.py:58: in run
    result = self.fn(*self.args, **self.kwargs)
argopy\stores\filesystems.py:419: in _mfprocessor
    ds = self.open_dataset(url, **open_dataset_opts)
argopy\stores\filesystems.py:400: in open_dataset
    with self.open(path) as of:
argopy\stores\filesystems.py:232: in open
    return self.fs.open(path, *args, **kwargs)
C:\Users\runneradmin\micromamba\envs\argopy-tests\Lib\site-packages\fsspec\implementations\cached.py:446: in <lambda>
    return lambda *args, **kw: getattr(type(self), item).__get__(self)(
C:\Users\runneradmin\micromamba\envs\argopy-tests\Lib\site-packages\fsspec\spec.py:1303: in open
    f = self._open(
C:\Users\runneradmin\micromamba\envs\argopy-tests\Lib\site-packages\fsspec\implementations\cached.py:446: in <lambda>
    return lambda *args, **kw: getattr(type(self), item).__get__(self)(
C:\Users\runneradmin\micromamba\envs\argopy-tests\Lib\site-packages\fsspec\implementations\cached.py:707: in _open
    self.save_cache()
C:\Users\runneradmin\micromamba\envs\argopy-tests\Lib\site-packages\fsspec\implementations\cached.py:446: in <lambda>
    return lambda *args, **kw: getattr(type(self), item).__get__(self)(
C:\Users\runneradmin\micromamba\envs\argopy-tests\Lib\site-packages\fsspec\implementations\cached.py:206: in save_cache
    self._metadata.save()
C:\Users\runneradmin\micromamba\envs\argopy-tests\Lib\site-packages\fsspec\implementations\cache_metadata.py:227: in save
    self._save(cache, fn)
C:\Users\runneradmin\micromamba\envs\argopy-tests\Lib\site-packages\fsspec\implementations\cache_metadata.py:75: in _save
    with atomic_write(fn, mode="w") as f:
C:\Users\runneradmin\micromamba\envs\argopy-tests\Lib\contextlib.py:144: in __exit__
    next(self.gen)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

path = 'C:\\Users\\RUNNER~1\\AppData\\Local\\Temp\\tmpcf4mglfi\\cache'
mode = 'w'

    @contextlib.contextmanager
    def atomic_write(path: str, mode: str = "wb"):
        """
        A context manager that opens a temporary file next to `path` and, on exit,
        replaces `path` with the temporary file, thereby updating `path`
        atomically.
        """
        fd, fn = tempfile.mkstemp(
            dir=os.path.dirname(path), prefix=os.path.basename(path) + "-"
        )
        try:
            with open(fd, mode) as fp:
                yield fp
        except BaseException:
            with contextlib.suppress(FileNotFoundError):
                os.unlink(fn)
            raise
        else:
>           os.replace(fn, path)
E           PermissionError: [WinError 5] Access is denied: 'C:\\Users\\RUNNER~1\\AppData\\Local\\Temp\\tmpcf4mglfi\\cache-gib2g7kt' -> 'C:\\Users\\RUNNER~1\\AppData\\Local\\Temp\\tmpcf4mglfi\\cache'

C:\Users\runneradmin\micromamba\envs\argopy-tests\Lib\site-packages\fsspec\utils.py:625: PermissionError

To view more test analytics, go to the Test Analytics Dashboard
📢 Thoughts on this report? Let us know!