Repository: intake/intake
Pull request #564 (2020-12-23 21:38:31)

Derived datasets
cc @philippjfr (this is not formulated in terms of param, but could be)


Patch:
diff --git a/docs/source/catalog.rst b/docs/source/catalog.rst
index 4cd51cc..03e08b6 100644
--- a/docs/source/catalog.rst
+++ b/docs/source/catalog.rst
@@ -284,7 +284,7 @@ Expressed in YAML, a parameter may look as follows:
     parameters:
       name:
         description: name to use  # human-readable text for what this parameter means
-        type: str  # one of bool, str, int, float, list[str], list[int], list[float], datetime
+        type: str  # optional, one of bool, str, int, float, list[str], list[int], list[float], datetime
         default: normal  # optional, value to assume if user does not override
         allowed: ["normal", "strange"]  # optional, list of values that are OK, for validation
         min: "n"  # optional, minimum allowed, for validation
diff --git a/docs/source/guide.rst b/docs/source/guide.rst
index e9895f9..8739629 100644
--- a/docs/source/guide.rst
+++ b/docs/source/guide.rst
@@ -14,3 +14,4 @@ how to use the graphical interface, plotting, etc.
     plotting.rst
     plugin-directory.rst
     server.rst
+    transforms.rst
diff --git a/docs/source/transforms.rst b/docs/source/transforms.rst
new file mode 100644
index 0000000..38f74cd
--- /dev/null
+++ b/docs/source/transforms.rst
@@ -0,0 +1,38 @@
+Dataset Transforms
+------------------
+
+aka. derived datasets.
+
+(experimental)
+
+Intake allows for the definition of data sources which take as their input
+another source in the same directory, so that you have the opportunity to
+present *processing* to the user of the catalog.
+
+Example
+~~~~~~~
+
+This example is taken from the Intake test suite.
+
+Text to come, watch this space...
+
+API
+~~~
+
+.. autosummary::
+   intake.source.derived.DerivedSource
+   intake.source.derived.Alias
+   intake.source.derived.GenericTransform
+   intake.source.derived.DataFrameTransform
+   intake.source.derived.Columns
+
+.. autoclass:: intake.source.derived.DerivedSource
+   :members:
+.. autoclass:: intake.source.derived.Alias
+   :members:
+.. autoclass:: intake.source.derived.GenericTransform
+   :members:
+.. autoclass:: ntake.source.derived.DataFrameTransform
+   :members:
+.. autoclass:: intake.source.derived.Columns
+   :members:
diff --git a/intake/catalog/local.py b/intake/catalog/local.py
index 262d5f9..36d56c2 100644
--- a/intake/catalog/local.py
+++ b/intake/catalog/local.py
@@ -779,18 +779,20 @@ class YAMLFilesCatalog(Catalog):
                 else:
                     # store a catalog entry
                     self._cats[f.path] = entry
+        entries = {}
         for name, entry in list(self._cats.items()):
             if self._flatten:
                 entry.reload()
-                inter = set(entry._entries).intersection(self._entries)
+                inter = set(entry._entries).intersection(entries)
                 if inter:
                     raise ValueError(
                         'Conflicting names when flattening multiple'
                         ' catalogs. Sources %s exist in more than'
                         ' one' % inter)
-                self._entries.update(entry._entries)
+                entries.update(entry._entries)
             else:
-                self._entries[entry._name] = entry
+                entries[entry._name] = entry
+        self._entries = entries
 
 
 class MergedCatalog(Catalog):
@@ -804,9 +806,9 @@ class MergedCatalog(Catalog):
     def _load(self):
         for catalog in self._catalogs:
             catalog._load()
-
-    def _make_entries_container(self):
-        return collections.ChainMap(*(catalog._entries for catalog in self._catalogs))
+        self._entries = collections.ChainMap(
+            *(catalog._entries for catalog in self._catalogs)
+        )
 
 
 class EntrypointEntry(CatalogEntry):
diff --git a/intake/source/base.py b/intake/source/base.py
index 9c6ce58..5a830a9 100644
--- a/intake/source/base.py
+++ b/intake/source/base.py
@@ -4,9 +4,10 @@
 #
 # The full license is in the LICENSE file, distributed with this software.
 #-----------------------------------------------------------------------------
-''' Base classes for Data Loader interface
+"""
+Base classes for Data Loader interface
+"""
 
-'''
 from yaml import dump
 
 from .cache import make_caches
@@ -209,7 +210,7 @@ class DataSourceBase(DictSerialiseMixin):
     def __init__(self, storage_options=None, metadata=None):
         # default data
         self.metadata = metadata or {}
-        if isinstance(self.metadata, dict):
+        if isinstance(self.metadata, dict) and storage_options is None:
             storage_options = self._captured_init_kwargs.get('storage_options',
                                                              {})
         self.storage_options = storage_options
@@ -471,75 +472,3 @@ class PatternMixin(object):
         elif self.path_as_pattern:
             return path_to_pattern(self._original_urlpath, self.metadata)
         return
-
-
-class AliasSource(DataSource):
-    """Refer to another named source, unmodified
-
-    The purpose of an Alias is to be able to refer to other source(s) in the
-    same catalog, perhaps leaving the choice of which target to load up to the
-    user. This source makes no sense outside of a catalog.
-
-    In this case, the output of the target source is not modified, but this
-    class acts as a prototype 'derived' source for processing the output of
-    some standard driver.
-
-    After initial discovery, the source's container and other details will be
-    updated from the target; initially, the AliasSource container is not
-    any standard.
-    """
-    container = 'other'
-    version = 1
-    name = 'alias'
-
-    def __init__(self, target, mapping=None, metadata=None, **kwargs):
-        """
-
-        Parameters
-        ----------
-        target: str
-            Name of the source to load, must be a key in the same catalog
-        mapping: dict or None
-            If given, use this to map the string passed as ``target`` to
-            entries in the catalog
-        metadata: dict or None
-            Extra metadata to associate
-        kwargs: passed on to the target
-        """
-        super(AliasSource, self).__init__(metadata)
-        self.target = target
-        self.mapping = mapping or {target: target}
-        self.kwargs = kwargs
-        self.metadata = metadata
-        self.source = None
-
-    def _get_source(self):
-        if self.cat is None:
-            raise ValueError('AliasSource cannot be used outside a catalog')
-        if self.source is None:
-            self.source = self.cat[self.mapping[self.target]](
-                metadata=self.metadata, **self.kwargs)
-            self.metadata = self.source.metadata.copy()
-            self.container = self.source.container
-            self.partition_access = self.source.partition_access
-            self.description = self.source.description
-
-    def discover(self):
-        self._get_source()
-        return self.source.discover()
-
-    def read(self):
-        self._get_source()
-        return self.source.read()
-
-    def read_partition(self, i):
-        self._get_source()
-        return self.source.read_partition(i)
-
-    def read_chunked(self):
-        self._get_source()
-        return self.source.read_chunked()
-
-    def to_dask(self):
-        self._get_source()
-        return self.source.to_dask()
diff --git a/intake/source/derived.py b/intake/source/derived.py
new file mode 100644
index 0000000..00b1f5b
--- /dev/null
+++ b/intake/source/derived.py
@@ -0,0 +1,267 @@
+from . import import_name
+from .base import DataSource, Schema
+
+
+class AliasSource(DataSource):
+    """Refer to another named source, unmodified
+
+    The purpose of an Alias is to be able to refer to other source(s) in the
+    same catalog, perhaps leaving the choice of which target to load up to the
+    user. This source makes no sense outside of a catalog.
+
+    In this case, the output of the target source is not modified, but this
+    class acts as a prototype 'derived' source for processing the output of
+    some standard driver.
+
+    After initial discovery, the source's container and other details will be
+    updated from the target; initially, the AliasSource container is not
+    any standard.
+    """
+    container = 'other'
+    version = 1
+    name = 'alias'
+
+    def __init__(self, target, mapping=None, metadata=None, **kwargs):
+        """
+
+        Parameters
+        ----------
+        target: str
+            Name of the source to load, must be a key in the same catalog
+        mapping: dict or None
+            If given, use this to map the string passed as ``target`` to
+            entries in the catalog
+        metadata: dict or None
+            Extra metadata to associate
+        kwargs: passed on to the target
+        """
+        super(AliasSource, self).__init__(metadata)
+        self.target = target
+        self.mapping = mapping or {target: target}
+        self.kwargs = kwargs
+        self.metadata = metadata
+        self.source = None
+
+    def _get_source(self):
+        if self.cat is None:
+            raise ValueError('AliasSource cannot be used outside a catalog')
+        if self.source is None:
+            self.source = self.cat[self.mapping[self.target]](
+                metadata=self.metadata, **self.kwargs)
+            self.metadata = self.source.metadata.copy()
+            self.container = self.source.container
+            self.partition_access = self.source.partition_access
+            self.description = self.source.description
+
+    def discover(self):
+        self._get_source()
+        return self.source.discover()
+
+    def read(self):
+        self._get_source()
+        return self.source.read()
+
+    def read_partition(self, i):
+        self._get_source()
+        return self.source.read_partition(i)
+
+    def read_chunked(self):
+        self._get_source()
+        return self.source.read_chunked()
+
+    def to_dask(self):
+        self._get_source()
+        return self.source.to_dask()
+
+
+def first(targets, cat, kwargs):
+    """A target chooser that simply picks the first from the given list
+
+    This is the default, particularly for the case of only one element in
+    the list
+    """
+    targ = targets[0]
+    if cat:
+        s = cat[targ]
+        if kwargs and targ in kwargs:
+            s = kwargs.configure_new(**kwargs[targ])
+        return s
+    else:  # pragma: no cover
+        # for testing only
+        return targ
+
+
+def first_discoverable(targets, cat, kwargs):
+    """A target chooser: the first target for which discover() succeeds
+
+    This may be useful where some drivers are not importable, or some
+    sources can be available only sometimes.
+    """
+    for t in targets:
+        try:
+            if cat:
+                s = cat[t]
+                if kwargs and t in kwargs:
+                    s = s.configure_new(**kwargs[t])
+            else:
+                s = t
+            s.discover()
+            return s
+        except Exception:
+            pass
+    raise RuntimeError("No targets succeeded at discover()")
+
+
+class DerivedSource(DataSource):
+    """Base source deriving from another source in the same catalog
+
+    Target picking and parameter validation are performed here, but
+    you probably want to subclass from one of the more specific
+    classes like ``DataFrameTransform``.
+    """
+
+    input_container = "other"  # no constraint
+    container = 'other'  # to be filled in per instance at access time
+    required_params = []  # list of kwargs that must be present
+    optional_params = {}  # optional kwargs with defaults
+
+    def __init__(self, targets, target_chooser=first, target_kwargs=None,
+                 container=None, metadata=None, **kwargs):
+        """
+
+        Parameters
+        ----------
+        targets: list of string or DataSources
+            If string(s), refer to entries of the same catalog as this Source
+        target_chooser: function to choose between targets
+            function(targets, cat) -> source, or a fully-qualified dotted string pointing
+            to it
+        target_kwargs: dict of dict with keys matching items of targets
+        container: str (optional)
+            Assumed output container, if known/different from input
+        """
+        self.targets = targets
+        self._chooser = (target_chooser if callable(target_chooser)
+                         else import_name(target_chooser))
+        self._kwargs = target_kwargs
+        self._source = None
+        self._params = kwargs
+        if container:
+            self.container = container
+        self._validate_params()
+        super().__init__(metadata=metadata)
+
+    def _validate_params(self):
+        """That all required params are present and that optional types match"""
+        assert set(self.required_params) - set(self._params) == set()
+        for par, val in self.optional_params.items():
+            if par not in self._params:
+                self._params[par] = val
+
+    def _pick(self):
+        """ Pick the source from the given targets """
+        self._source = self._chooser(self.targets, self.cat, self._kwargs)
+        if self.input_container != "other":
+            assert self._source.container == self.input_container
+
+        self.metadata['target'] = self._source.metadata
+        if self.container is None:
+            self.container = self._source.container
+
+
+class GenericTransform(DerivedSource):
+    required_params = ["transform", "transform_kwargs"]
+    optional_params = {"allow_dask": True}
+    """
+    Perform an arbitrary function to transform an input
+    
+        transform: function to perform transform
+            function(container_object) -> output, or a fully-qualified dotted string pointing
+            to it
+        transform_params: dict
+            The keys are names of kwargs to pass to the transform function. Values are either
+            concrete values to pass; or param objects which can be made into widgets (but
+            must have a default value) - or a spec to be able to make these objects.
+        allow_dask: bool (optional, default True)
+            Whether to_dask() is expected to work, which will in turn call the
+            target's to_dasK()
+    """
+
+    def _validate_params(self):
+        super()._validate_params()
+        transform = self._params["transform"]
+        self._transform = (transform if callable(transform)
+                           else import_name(transform))
+
+    def _get_schema(self):
+        """We do not know the schema of a generic transform"""
+        self._pick()
+        return Schema()
+
+    def to_dask(self):
+        if not self._params['allow_dask']:
+            raise ValueError("This transform is not compatible with Dask"
+                             "because it has use_dask=False")
+        return self._transform(self._source.to_dask(), **self._params["transform_kwargs"])
+
+    def read(self):
+        return self._transform(self._source.read(), **self._params["transform_kwargs"])
+
+
+class DataFrameTransform(GenericTransform):
+    """Transform where the input and output are both Dask-compatible dataframes
+
+    This derives from GenericTransform, and you mus supply ``transform`` and
+    any ``transform_kwargs``.
+    """
+
+    input_container = "dataframe"
+    container = "dataframe"
+    optional_params = {}
+    _df = None
+
+    def to_dask(self):
+        if self._df is None:
+            self._pick()
+            self._df = self._transform(self._source.to_dask(),
+                                       **self._params["transform_kwargs"])
+        return self._df
+
+    def _get_schema(self):
+        """load metadata only if needed"""
+        self.to_dask()
+        return Schema(dtype=self._df.dtypes,
+                      shape=(None, len(self._df.columns)),
+                      npartitions=self._df.npartitions,
+                      metadata=self.metadata)
+
+    def read(self):
+        return self.to_dask().compute()
+
+
+class Columns(DataFrameTransform):
+    """Simple dataframe transform to pick columns
+
+    Given as an example of how to make a specific dataframe transform.
+    Note that you could use DataFrameTransform directly, by writing a
+    function to choose the columns instead of a method as here.
+    """
+
+    input_container = "dataframe"
+    container = "dataframe"
+    required_params = ["columns"]
+
+    def __init__(self, columns, **kwargs):
+        """
+            columns: list of labels (usually str) or slice
+                Columns to choose from the target dataframe
+        """
+        # this class wants requires "columns", but DataFrameTransform
+        # uses "transform_kwargs", which we don't need since we use a method for the
+        # transform
+        kwargs.update(transform=self.pick_columns, columns=columns,
+                      transform_kwargs={})
+        super().__init__(**kwargs)
+
+    def pick_columns(self, df):
+        return df[self._params["columns"]]
diff --git a/setup.py b/setup.py
index 91e0baa..a569f22 100644
--- a/setup.py
+++ b/setup.py
@@ -53,6 +53,7 @@ setup(
             'numpy = intake.source.npy:NPySource',
             'ndzarr = intake.source.zarr:ZarrArraySource',
             'zarr_cat = intake.catalog.zarr:ZarrGroupCatalog',
+            'alias = intake.source.derived:AliasSource',
         ]
     },
     classifiers=[
@@ -60,6 +61,7 @@ setup(
         "Programming Language :: Python :: 3.6",
         "Programming Language :: Python :: 3.7",
         "Programming Language :: Python :: 3.8",
+        "Programming Language :: Python :: 3.9",
     ],
     python_requires=">=3.6",
     long_description=open('README.md').read(),


Test patch:
diff --git a/intake/catalog/tests/catalog_alias.yml b/intake/catalog/tests/catalog_alias.yml
index a9c66eb..1e7582d 100644
--- a/intake/catalog/tests/catalog_alias.yml
+++ b/intake/catalog/tests/catalog_alias.yml
@@ -1,6 +1,6 @@
 sources:
-  test_cache:
-    description: cache a csv file from the local filesystem
+  input_data:
+    description: a local data file
     driver: csv
     args:
       urlpath: '{{ CATALOG_DIR }}cache_data/states.csv'
@@ -11,15 +11,15 @@ sources:
       path: "{{ CATALOG_DIR }}/small.npy"
       chunks: 5
   alias0:
-    driver: intake.source.base.AliasSource
+    driver: intake.source.derived.AliasSource
     args:
-      target: test_cache
+      target: input_data
   alias1:
-    driver: intake.source.base.AliasSource
+    driver: alias
     args:
       target: "{{choice}}"
       mapping:
-        first: test_cache
+        first: input_data
         second: arr_cache
     parameters:
       choice:
@@ -27,3 +27,18 @@ sources:
         type: str
         default: first
         allowed: ["first", "second"]
+  derive_cols:
+    driver: intake.source.derived.Columns
+    args:
+      targets:
+        - input_data
+      columns: ["state", "slug"]
+  derive_cols_func:
+    driver: intake.source.derived.DataFrameTransform
+    args:
+      targets:
+        - input_data
+      transform: "intake.source.tests.test_derived._pick_columns"
+      transform_kwargs:
+        columns: ["state", "slug"]
+
diff --git a/intake/source/tests/alias.yaml b/intake/source/tests/alias.yaml
index 9d95904..e424db1 100644
--- a/intake/source/tests/alias.yaml
+++ b/intake/source/tests/alias.yaml
@@ -8,13 +8,13 @@ sources:
     args:
       urlpath: '{{ CATALOG_DIR }}/*.yaml'
   alias1:
-    driver: intake.source.base.AliasSource
+    driver: intake.source.derived.AliasSource
     args:
       mapping:
         first: csvs
         second: yamls
       target: first
   alias2:
-    driver: intake.source.base.AliasSource
+    driver: intake.source.derived.AliasSource
     args:
       target: csvs
diff --git a/intake/source/tests/der.yaml b/intake/source/tests/der.yaml
new file mode 100644
index 0000000..1f95db8
--- /dev/null
+++ b/intake/source/tests/der.yaml
@@ -0,0 +1,13 @@
+sources:
+  base:
+    driver: csv
+    args:
+      urlpath: "{{CATALOG_DIR}}/sample1.csv"
+  cols:
+    driver: intake.source.derived.Columns
+    args:
+      targets:
+        - base
+      columns:
+        - score
+        - rank
diff --git a/intake/source/tests/test_base.py b/intake/source/tests/test_base.py
index e214416..6f17418 100644
--- a/intake/source/tests/test_base.py
+++ b/intake/source/tests/test_base.py
@@ -13,6 +13,7 @@ import numpy as np
 import pandas as pd
 
 import intake.source.base as base
+import intake.source.derived as der
 
 
 def test_datasource_base_method_exceptions():
@@ -310,7 +311,7 @@ def test_yaml_method(source_python):
 
 
 def test_alias_fail():
-    s = base.AliasSource('atarget')
+    s = der.AliasSource('atarget')
     s.container == 'other'
     with pytest.raises(ValueError):
         s.read()
diff --git a/intake/source/tests/test_derived.py b/intake/source/tests/test_derived.py
new file mode 100644
index 0000000..da825e7
--- /dev/null
+++ b/intake/source/tests/test_derived.py
@@ -0,0 +1,23 @@
+import os
+import intake
+
+catfile = os.path.join(os.path.dirname(__file__), "..", "..",
+                       "catalog", "tests", "catalog_alias.yml")
+
+
+def test_columns():
+    cat = intake.open_catalog(catfile)
+    df1 = cat.input_data.read()
+    df2 = cat.derive_cols.read()
+    assert df1[["state", "slug"]].equals(df2)
+
+
+def _pick_columns(df, columns):
+    return df[columns]
+
+
+def test_df_transform():
+    cat = intake.open_catalog(catfile)
+    df1 = cat.input_data.read()
+    df2 = cat.derive_cols_func.read()
+    assert df1[["state", "slug"]].equals(df2)