Rewrite for intake2 (#154)

* Rewrite for intake2

Author: martindurant

.github/workflows/ci.yaml

@@ -30,16 +30,10 @@ jobs:
       - uses: actions/checkout@v4
 
       - name: set up conda environment
-        uses: mamba-org/setup-micromamba@v1
+        uses: conda-incubator/setup-miniconda@v3
         with:
           environment-file: ci/environment.yml
-          init-shell: >-
-            bash
-          cache-environment: true
-          cache-downloads: true
-          post-cleanup: "all"
-          create-args: |
-            python=${{ matrix.python-version }}
+          python-version: ${{ matrix.python-version }}
 
       - name: Install intake-thredds
         run: |

.github/workflows/nightly.yaml

@@ -22,16 +22,10 @@ jobs:
       - uses: actions/checkout@v4
 
       - name: set up conda environment
-        uses: mamba-org/setup-micromamba@v1
+        uses: conda-incubator/setup-miniconda@v3
         with:
           environment-file: ci/environment-upstream-dev.yml
-          init-shell: >-
-            bash
-          cache-environment: true
-          cache-downloads: true
-          post-cleanup: "all"
-          create-args: |
-            python=3.12
+          python-version: 3.12
 
       - name: Install intake-thredds
         run: |

intake_thredds/cat.py

@@ -1,104 +1,45 @@
-from intake.catalog import Catalog
-from intake.catalog.local import LocalCatalogEntry
-
-
-class ThreddsCatalog(Catalog):
-    """Intake catalog interface to a thredds catalog.
-
-    Parameters
-    ----------
-    url : str
-        Location of thredds catalog.
-    driver : str
-        Select driver to access data. Choose from 'netcdf' and 'opendap'.
-    intake_xarray_kwargs : dict
-        Keyword arguments to pass to intake_xarray DataSource.
-    **kwargs :
-        Additional keyword arguments are passed through to the
-        :py:class:`~intake.catalog.Catalog` base class.
-
-    Examples
-    --------
-    >>> import intake
-    >>> cat_url = 'https://psl.noaa.gov/thredds/catalog/Datasets/noaa.ersst/catalog.xml'
-    >>> cat = intake.open_thredds_cat(cat_url)
-    """
-
-    name = 'thredds_cat'
-
-    def __init__(self, url: str, driver: str = 'opendap', intake_xarray_kwargs=None, **kwargs):
-        self.url = url
-        self.driver = driver
-        self.intake_xarray_kwargs = intake_xarray_kwargs or {'chunks': {}}
-        super().__init__(**kwargs)
-
-    def _load(self):
-        from siphon.catalog import TDSCatalog
-
-        if 'simplecache::' in self.url:
-            if self.driver == 'netcdf':
-                self.cache = True
-                self.url_no_simplecache = self.url.replace('simplecache::', '')
-                self.metadata.update({'fsspec_pre_url': 'simplecache::'})
-            else:
-                raise ValueError(
-                    f'simplecache requires driver="netcdf", found driver="{self.driver}".'
-                )
-        else:
-            self.cache = False
-            self.url_no_simplecache = self.url
-
-        self.cat = TDSCatalog(self.url_no_simplecache)
-        if self.name is None:
-            self.name = self.cat.catalog_name
-        self.metadata.update(self.cat.metadata)
-
-        # sub-cats
-        self._entries = {
-            r.title: LocalCatalogEntry(
-                r.title,
-                'THREDDS cat',
-                'thredds_cat',
-                True,
-                {'url': r.href},
-                [],
-                [],
-                self.metadata,
-                None,
-                catalog=self,
-            )
-            for r in self.cat.catalog_refs.values()
-        }
-
-        def access_urls(ds, self):
-            if self.driver == 'opendap':
-                driver_for_access_urls = 'OPENDAP'
-            elif self.driver == 'netcdf':
-                driver_for_access_urls = 'HTTPServer'
-            url = ds.access_urls[driver_for_access_urls]
-            if 'fsspec_pre_url' in self.metadata.keys():
-                url = f'{self.metadata["fsspec_pre_url"]}{url}'
-            return url
-
-        def _update_args(ds):
-            args = self.intake_xarray_kwargs.copy()
-            args.update({'urlpath': access_urls(ds, self)})
-            return args
-
-        self._entries.update(
-            {
-                ds.name: LocalCatalogEntry(
-                    ds.name,
-                    'THREDDS data',
-                    self.driver,
-                    True,
-                    _update_args(ds),
-                    [],
-                    [],
-                    {},
-                    None,
-                    catalog=self,
-                )
-                for ds in self.cat.datasets.values()
-            }
-        )
+from intake.readers import Service
+from intake.readers.catalogs import THREDDSCatalogReader
+
+
+class ThreddsCatalog:
+    """Intake catalog interface to a thredds catalog."""
+
+    def __new__(cls, url: str, driver: str = 'opendap', intake_xarray_kwargs=None, metadata=None):
+        """
+        Parameters
+        ----------
+        url : str
+            Location of thredds catalog.
+        driver : str
+            Select driver to access data. Choose from 'netcdf' and 'opendap'.
+        intake_xarray_kwargs : dict
+            Keyword arguments to pass to intake_xarray DataSource.
+        **kwargs :
+            Additional keyword arguments are passed through to the
+            :py:class:`~intake.catalog.Catalog` base class.
+
+        Examples
+        --------
+        >>> import intake
+        >>> cat_url = 'https://psl.noaa.gov/thredds/catalog/Datasets/noaa.ersst/catalog.xml'
+        >>> cat = intake.open_thredds_cat(cat_url)
+        """
+
+        simplecache = url.startswith('simplecache:')
+        if simplecache and driver == 'opendap':
+            raise ValueError('simplecache requires driver="netcdf"')
+        url = url.removeprefix('simplecache::')
+        data = Service(url)
+        reader = THREDDSCatalogReader(data, make=driver[-3:])
+        cat = reader.read()
+        if metadata:
+            cat.metadata.update(metadata)
+        if simplecache:
+            for d in cat.data.values():
+                d.kwargs['url'] = 'simplecache::' + d.kwargs['url']
+        if intake_xarray_kwargs:
+            intake_xarray_kwargs.update(intake_xarray_kwargs.pop('xarray_kwargs', {}))
+            for d in cat.entries.values():
+                d.kwargs.update(intake_xarray_kwargs)
+        return cat

intake_thredds/source.py

@@ -1,12 +1,9 @@
 import fnmatch
 
-from intake_xarray.base import DataSourceMixin
-from tqdm.auto import tqdm
-
 from .cat import ThreddsCatalog
 
 
-class THREDDSMergedSource(DataSourceMixin):
+class THREDDSMergedSource:
     """Merges multiple datasets into a single datasets.
 
     This source takes a THREDDS URL and a path to descend down, and calls the
@@ -50,21 +47,17 @@ class THREDDSMergedSource(DataSourceMixin):
 
     """
 
-    version = '1.0'
-    container = 'xarray'
-    name = 'thredds_merged'
-    partition_access = True
-
     def __init__(
         self,
         url,
         path,
         driver='opendap',
-        xarray_kwargs={},
+        xarray_kwargs=None,
         concat_kwargs=None,
         metadata=None,
     ):
-        super().__init__(metadata=metadata)
+        xarray_kwargs = xarray_kwargs or {}
+        self.metadata = metadata or {}
         self.urlpath = url
         if 'simplecache::' in url:
             self.metadata.update({'fsspec_pre_url': 'simplecache::'})
@@ -78,33 +71,34 @@ def __init__(
         self.driver = driver
         self.xarray_kwargs = xarray_kwargs
         self.concat_kwargs = concat_kwargs
-        self._ds = None
 
-    def _open_dataset(self):
+    def read(self, xarray_kwargs=None):
         import xarray as xr
+        from tqdm import tqdm
 
-        if self._ds is None:
-            cat = ThreddsCatalog(self.urlpath, driver=self.driver)
-            for i in range(len(self.path)):
-                part = self.path[i]
-                if '*' not in part and '?' not in part:
-                    cat = cat[part](driver=self.driver)
-                else:
-                    break
-            path = self.path[i:]
-            data = [
-                ds(xarray_kwargs=self.xarray_kwargs).to_dask()
-                for ds in tqdm(_match(cat, path), desc='Dataset(s)', ncols=79)
-            ]
-            if self.concat_kwargs:
-                self._ds = xr.concat(data, **self.concat_kwargs)
+        cat = ThreddsCatalog(self.urlpath, driver=self.driver, metadata=self.metadata)
+        for i in range(len(self.path)):
+            part = self.path[i]
+            if '*' not in part and '?' not in part:
+                cat = cat[part].read(make=self.driver[-3:])
             else:
-                self._ds = xr.combine_by_coords(data, combine_attrs='override')
+                break
+        path = self.path[i:]
+        data = [
+            ds(**self.xarray_kwargs).read()
+            for ds in tqdm(_match(cat, path), desc='Dataset(s)', ncols=79)
+        ]
+        if self.concat_kwargs:
+            return xr.concat(data, **self.concat_kwargs)
+        else:
+            return xr.combine_by_coords(data, combine_attrs='override')
+
+    to_dask = read
 
 
 def _match(cat, patterns):
     out = []
-    for name in cat:
+    for name in cat.entries:
         if fnmatch.fnmatch(name, patterns[0]):
             if len(patterns) == 1:
                 out.append(cat[name](chunks={}))

requirements.txt

@@ -1,8 +1,8 @@
 aiohttp >=3.7
 fsspec >=0.8.5
 h5netcdf >=0.8.1
-intake-xarray >=0.3
-intake>=0.6.6,<2
+intake-xarray
+intake>2
 pydap
 siphon
 tqdm

tests/test_cat.py

@@ -1,5 +1,4 @@
 import intake
-import intake_xarray
 import pytest
 import xarray as xr
 
@@ -13,9 +12,8 @@ def thredds_cat_url():
 def test_ThreddsCatalog_init_catalog(thredds_cat_url):
     """Test initialization of ThreddsCatalog."""
     cat = intake.open_thredds_cat(thredds_cat_url)
-    assert isinstance(cat, intake.catalog.Catalog)
-    assert cat.metadata == cat.cat.metadata
-    assert cat.discover()['container'] == 'catalog'
+    assert isinstance(cat, intake.Catalog)
+    assert cat.metadata
 
     assert 'err.mnmean.v3.nc' in cat
 
@@ -28,42 +26,32 @@ def test_ThreddsCatalog(thredds_cat_url, driver):
     """Test entry.to_dask() is xr.Dataset and allows opendap and netcdf as source."""
     cat = intake.open_thredds_cat(thredds_cat_url, driver=driver)
     entry = cat['sst.mon.19712000.ltm.v3.nc']
-    if driver == 'opendap':
-        assert isinstance(entry, intake_xarray.opendap.OpenDapSource)
-    elif driver == 'netcdf':
-        assert isinstance(entry, intake_xarray.netcdf.NetCDFSource)
-    d = entry.describe()
-    assert d['name'] == 'sst.mon.19712000.ltm.v3.nc'
-    assert d['container'] == 'xarray'
-    assert d['plugin'] == [driver]
+    assert isinstance(entry, intake.readers.XArrayDatasetReader)
     if driver == 'opendap':
         loc = 'dodsC'
     elif driver == 'netcdf':
         loc = 'fileServer'
     assert (
-        d['args']['urlpath']
+        entry.data.url
         == f'https://psl.noaa.gov/thredds/{loc}/Datasets/noaa.ersst/sst.mon.19712000.ltm.v3.nc'
     )
-    ds = entry(chunks={}).to_dask()
+    ds = entry.read()
     assert isinstance(ds, xr.Dataset)
 
 
-def test_ThreddsCatalog_simplecache_netcdf(thredds_cat_url):
+def test_ThreddsCatalog_simplecache_netcdf(thredds_cat_url, tmpdir):
     """Test that ThreddsCatalog allows simplecache:: in url if netcdf as source."""
     import os
 
     import fsspec
 
-    fsspec.config.conf['simplecache'] = {'cache_storage': 'my_caching_folder', 'same_names': True}
+    fsspec.config.conf['simplecache'] = {'cache_storage': str(tmpdir), 'same_names': True}
     cat = intake.open_thredds_cat(f'simplecache::{thredds_cat_url}', driver='netcdf')
     entry = cat['sst.mon.19712000.ltm.v3.nc']
-    ds = entry(chunks={}).to_dask()
+    ds = entry().read()
     assert isinstance(ds, xr.Dataset)
-    # test files present
-    cached_file = 'my_caching_folder/sst.mon.19712000.ltm.v3.nc'
+    cached_file = f'{tmpdir}/sst.mon.19712000.ltm.v3.nc'
     assert os.path.exists(cached_file)
-    os.remove(cached_file)
-    assert not os.path.exists(cached_file)
 
 
 def test_ThreddsCatalog_simplecache_fails_opendap(thredds_cat_url):
@@ -83,5 +71,5 @@ def test_ThreddsCatalog_intake_xarray_kwargs():
         },
     )
     entry = cat['sst.mon.19712000.ltm.v3.nc']
-    ds = entry(chunks={}).to_dask()
+    ds = entry(chunks={}).read()
     assert isinstance(ds, xr.Dataset)