fix/plugin mechanism

docs/cli.rst

@@ -345,7 +345,7 @@ Generating small synthetic datasets to try it on:
 
     # JoinByTimestamp is a local plugin transformer (see Plugins below) - needed both to
     # build the pipeline above and to load it back for damast process
-    export DAMAST_PLUGIN_PATH=docs/examples/plugins
+    export DAMAST_PLUGIN_PATH=ais_osint=docs/examples/plugins
     damast process --pipeline pipelines/osint_ais_preparation.damast.ppl \
         --input-data df=docs/examples/data/ais.parquet \
         --input-data osint_events=docs/examples/data/osint.parquet \
@@ -365,7 +365,7 @@ resolvable, i.e., ``DAMAST_PLUGIN_PATH`` must be set for a local plugin, while i
 
 ::
 
-    export DAMAST_PLUGIN_PATH=docs/examples/plugins
+    export DAMAST_PLUGIN_PATH=ais_osint=docs/examples/plugins
     damast process --pipeline pipelines/osint_ais_preparation.damast.ppl --describe
 
 .. highlight:: none
@@ -624,20 +624,35 @@ registering such plugin transformers are supported - see :class:`damast.core.tra
 for the full API:
 
 - installable packages that declare their :class:`damast.core.transformations.PipelineElement` subclasses via the
-  ``damast.transformers`` entry-point group in their own ``pyproject.toml``::
+  ``damast.transformers`` entry-point group in their own ``pyproject.toml`` - either one entry per class, or
+  one entry per module, which registers every transformer defined in that module (or, for a package, in its
+  top-level submodules)::
 
       [project.entry-points."damast.transformers"]
       MyTransformer = "acme_pkg.transformers:MyTransformer"
+      acme_pkg = "acme_pkg.transformers"
 
-- local, ad-hoc ``*.py`` files that are not part of any installed package, made discoverable by
-  pointing the ``DAMAST_PLUGIN_PATH`` environment variable at the directory (or directories,
-  separated with ``os.pathsep``) that contains them
+- local directories that are not part of any installed package, made discoverable via the
+  ``DAMAST_PLUGIN_PATH`` environment variable (several entries separated with ``os.pathsep``). An entry
+  ``name=path`` loads the directory as a package called ``name``: its top-level ``*.py`` files become
+  ``name.<file>``, and may use relative imports (``from .helpers import x``), also into subdirectories.
+  The same can be done in code via ``plugin_manager.register_plugin_package(name, path)``.
+  Choose a distinctive name - one that is already importable is rejected. A bare ``path`` entry
+  (deprecated) loads each file as a flat module named after the file, without relative imports.
+
+The package name of a local directory is recorded in saved pipelines (e.g.
+``module_name: my_plugins.my_transformers``), so a pipeline can only be replayed with the directory
+registered under the same name - or with an installed package of that name, e.g. once the directory has
+been turned into an installable package with a module entry-point.
 
 Regardless of which of the two a transformer comes from, it is resolvable in code the same way,
-via the ``damast.plugins`` namespace::
+via ``damast.plugins.<package>`` - where ``<package>`` is the top-level package of the module
+defining it, i.e. ``acme_pkg`` for ``acme_pkg.transformers:MyTransformer``, or the name a local
+directory was registered under::
 
-    from damast.plugins import MyTransformer
+    from damast.plugins.acme_pkg import MyTransformer
 
+Scoping by package means two plugins can provide a transformer of the same name without clashing.
 ``damast.plugins`` resolves names lazily on first access, so nothing beyond the requested class
 is ever imported - see :mod:`damast.plugins` for details.
 
@@ -650,27 +665,29 @@ subclass in a loose ``*.py`` file, written like any other transformer:
 .. literalinclude:: ./examples/plugins/my_transformers.py
    :language: Python
 
-With ``DAMAST_PLUGIN_PATH`` pointing at the directory containing that file, ``damast plugins``
-lists it without requiring any further Python code:
+With ``DAMAST_PLUGIN_PATH`` registering the directory containing that file (here as package
+``my_plugins``), ``damast plugins`` lists it without requiring any further Python code:
 
 .. highlight:: none
 
 ::
 
-    $ export DAMAST_PLUGIN_PATH=./examples/plugins
+    $ export DAMAST_PLUGIN_PATH=my_plugins=./examples/plugins
     $ damast plugins
 
-    MyTripler: my_transformers:MyTripler
+    my_plugins (local: examples/plugins)
+        JoinByTimestamp  .osint_ais_transformers
+        MyTripler        .my_transformers
 
-``MyTripler`` is now resolvable via ``damast.plugins`` and can be used in a pipeline like any
-other transformer:
+``MyTripler`` is now resolvable via ``damast.plugins.my_plugins`` and can be used in a pipeline
+like any other transformer:
 
 .. literalinclude:: ./examples/damast-plugin-pipeline.py
    :language: Python
 
 The resulting pipeline can be applied like any other, e.g. via ``damast process`` (see `Process`_
-above), as long as ``DAMAST_PLUGIN_PATH`` is still set to a directory containing
-``my_transformers.py``:
+above), as long as ``DAMAST_PLUGIN_PATH`` still registers the directory containing
+``my_transformers.py`` as ``my_plugins``:
 
 .. highlight:: none
 
@@ -679,7 +696,7 @@ above), as long as ``DAMAST_PLUGIN_PATH`` is still set to a directory containing
     damast process --input-data data.parquet --pipeline pipelines/my-plugin-pipeline.damast.ppl
 
 Pipelines saved with a plugin transformer record where it came from under ``requires`` (the
-installed distribution and version, or the original local file path), so that loading the pipeline
-elsewhere fails with an actionable message - naming the missing package to ``pip install``, or the
-``DAMAST_PLUGIN_PATH`` directory to add - instead of a bare import error.
+installed distribution and version, or the local package name and directory), so that loading the
+pipeline elsewhere fails with an actionable message - naming the missing package to ``pip install``, or
+the ``DAMAST_PLUGIN_PATH`` entry to add - instead of a bare import error.
 

docs/examples/damast-osint-ais-pipeline.py

@@ -16,10 +16,10 @@
 
 # JoinByTimestamp lives in its own package/file in a real project - see docs/cli.rst >
 # Plugins. Here it's a local plugin file, resolved the same way.
-os.environ["DAMAST_PLUGIN_PATH"] = str(Path(__file__).parent / "plugins")
+os.environ["DAMAST_PLUGIN_PATH"] = f"ais_osint={Path(__file__).parent / 'plugins'}"
 
 from damast.core.dataprocessing import DataProcessingPipeline
-from damast.plugins import JoinByTimestamp
+from damast.plugins.ais_osint import JoinByTimestamp
 
 pipeline = DataProcessingPipeline(name="osint_ais_preparation",
                                   description="Join AIS pings with OSINT events by timestamp",

docs/examples/damast-plugin-pipeline.py

@@ -1,12 +1,13 @@
 import os
 from pathlib import Path
 
-# DAMAST_PLUGIN_PATH points at the directory holding my_transformers.py, so that
-# 'MyTripler' becomes resolvable via damast.plugins - see docs/examples/plugins/.
-os.environ["DAMAST_PLUGIN_PATH"] = str(Path(__file__).parent / "plugins")
+# DAMAST_PLUGIN_PATH registers the directory holding my_transformers.py as package
+# 'my_plugins', so that 'MyTripler' becomes resolvable via damast.plugins.my_plugins - see
+# docs/examples/plugins/.
+os.environ["DAMAST_PLUGIN_PATH"] = f"my_plugins={Path(__file__).parent / 'plugins'}"
 
 from damast.core import DataProcessingPipeline
-from damast.plugins import MyTripler
+from damast.plugins.my_plugins import MyTripler
 
 pipeline = DataProcessingPipeline(name="my-plugin-pipeline", base_dir=".")
 pipeline.add("Triple mmsi",

docs/usage.ipynb

@@ -337,30 +337,34 @@
     "Python package at all. Damast supports two ways to use `PipelineElement` classes that live\n",
     "outside of `damast` itself, so that pipelines using them stay reproducible when shared with\n",
     "others. Regardless of which of the two a transformer comes from, it is always resolvable the\n",
-    "same way in code - `from damast.plugins import MyTransformer` - without needing to import (or\n",
-    "even know) the underlying module or package that defines it.\n",
+    "same way in code - `from damast.plugins.<package> import MyTransformer`, where `<package>` is\n",
+    "the plugin's top-level package - without needing to import (or even know) the underlying module\n",
+    "that defines it. Scoping by package means two plugins can provide a transformer of the same name\n",
+    "without clashing.\n",
     "\n",
     "### Local, ad-hoc transformers via `DAMAST_PLUGIN_PATH`\n",
     "\n",
     "For a quick transformer that does not warrant its own package, put it in a plain `*.py` file and\n",
-    "point the `DAMAST_PLUGIN_PATH` environment variable at the directory containing it (multiple\n",
-    "directories can be separated with `os.pathsep`, just like `PATH`). Every top-level file found\n",
-    "there is imported once - using its filename stem as the module name - so any `PipelineElement`\n",
-    "subclasses it defines become resolvable exactly like classes from an installed package."
+    "register the directory containing it under a package name via the `DAMAST_PLUGIN_PATH`\n",
+    "environment variable, as `name=path` (multiple entries can be separated with `os.pathsep`, just\n",
+    "like `PATH`). Every top-level file found there is imported once as `name.<file>` - so files may\n",
+    "use relative imports among each other - and any `PipelineElement` subclasses they define become\n",
+    "resolvable exactly like classes from an installed package. Choose a distinctive name: it is\n",
+    "recorded in saved pipelines, and a name that is already importable is rejected."
    ]
   },
   {
    "cell_type": "code",
    "execution_count": null,
    "metadata": {},
    "outputs": [],
-   "source": "import os\nimport tempfile\nfrom pathlib import Path\n\nplugin_dir = Path(tempfile.mkdtemp())\n(plugin_dir / \"my_transformers.py\").write_text(\"\"\"\nimport polars\n\nfrom damast.core.dataframe import AnnotatedDataFrame\nfrom damast.core.decorators import describe, input, output\nfrom damast.core.transformations import PipelineElement\n\n\nclass Doubler(PipelineElement):\n    @describe(\"Doubles a column\")\n    @input({\"x\": {}})\n    @output({\"{{x}}_doubled\": {}})\n    def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame:\n        feature = self.get_name(\"x\")\n        df.lazyframe = df.lazyframe.with_columns(\n            (polars.col(feature) * 2).alias(f\"{feature}_doubled\")\n        )\n        return df\n\"\"\")\n\nos.environ[\"DAMAST_PLUGIN_PATH\"] = str(plugin_dir)\n\nfrom damast.core.transformations import PipelineElement  # noqa: E402\n\n# directories are only scanned once per process, so force a rescan after\n# changing DAMAST_PLUGIN_PATH or adding/editing files\nPipelineElement.reload_plugins()\nPipelineElement.list_plugins()"
+   "source": "import os\nimport tempfile\nfrom pathlib import Path\n\nplugin_dir = Path(tempfile.mkdtemp())\n(plugin_dir / \"my_transformers.py\").write_text(\"\"\"\nimport polars\n\nfrom damast.core.dataframe import AnnotatedDataFrame\nfrom damast.core.decorators import describe, input, output\nfrom damast.core.transformations import PipelineElement\n\n\nclass Doubler(PipelineElement):\n    @describe(\"Doubles a column\")\n    @input({\"x\": {}})\n    @output({\"{{x}}_doubled\": {}})\n    def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame:\n        feature = self.get_name(\"x\")\n        df.lazyframe = df.lazyframe.with_columns(\n            (polars.col(feature) * 2).alias(f\"{feature}_doubled\")\n        )\n        return df\n\"\"\")\n\nos.environ[\"DAMAST_PLUGIN_PATH\"] = f\"my_plugins={plugin_dir}\"\n\nfrom damast.core.transformations import PipelineElement  # noqa: E402\n\n# directories are only scanned once per process, so force a rescan after\n# changing DAMAST_PLUGIN_PATH or adding/editing files\nPipelineElement.reload_plugins()\nPipelineElement.list_plugins()"
   },
   {
    "cell_type": "markdown",
    "metadata": {},
    "source": [
-    "`Doubler` is now resolvable via `damast.plugins`, and can be used in a pipeline like any other\n",
+    "`Doubler` is now resolvable via `damast.plugins.my_plugins`, and can be used in a pipeline like any other\n",
     "transformer. `damast.plugins` resolves names lazily on first access - nothing beyond the\n",
     "requested class is imported, and it works no matter which of the two plugin sources defines\n",
     "it. Saving the pipeline records where the transformer came from, under `requires`, so a pipeline\n",
@@ -375,7 +379,7 @@
    "outputs": [],
    "source": [
     "from damast.core.dataprocessing import DataProcessingPipeline\n",
-    "from damast.plugins import (\n",
+    "from damast.plugins.my_plugins import (\n",
     "    Doubler,  # resolved from the file we just wrote to `plugin_dir`\n",
     ")\n",
     "\n",
@@ -401,12 +405,16 @@
     "MyTransformer = \"acme_pkg.transformers:MyTransformer\"\n",
     "```\n",
     "\n",
+    "Instead of one entry per class, an entry may also name a whole module, e.g.\n",
+    "`acme_pkg = \"acme_pkg.transformers\"`: every `PipelineElement` defined in that module - or, for a\n",
+    "package, in its top-level submodules - is then registered.\n",
+    "\n",
     "Once installed, it is discovered the same way as a local plugin - `PipelineElement.list_plugins()`\n",
     "merges both sources - and pipelines saved with it additionally record the distribution name and\n",
     "version under `requires`, so a version mismatch on reload is logged as a warning rather than\n",
     "silently changing behavior. Run `damast plugins` from the command line to list everything that is\n",
     "currently discoverable, from either source, without writing any Python. In code, it resolves the\n",
-    "same way as a local plugin too - `from damast.plugins import MyTransformer`."
+    "same way as a local plugin too - `from damast.plugins.acme_pkg import MyTransformer`."
    ]
   },
   {

pyproject.toml

@@ -71,7 +71,8 @@ repository = "https://github.com/simula/damast"
 # Third-party packages can register PipelineElement subclasses as damast plugins via
 # their own pyproject.toml, e.g.:
 #   [project.entry-points."damast.transformers"]
-#   MyTransformer = "acme_pkg.transformers:MyTransformer"
+#   MyTransformer = "acme_pkg.transformers:MyTransformer"   # a single class
+#   acme_pkg = "acme_pkg.transformers"                      # all classes of a module/package
 # Discoverable via `damast plugins` or PipelineElement.list_plugins().
 
 [project.optional-dependencies]

src/damast/cli/plugins.py

@@ -2,7 +2,7 @@
 from argparse import ArgumentParser
 
 from damast.cli.base import BaseParser
-from damast.core.transformations import PipelineElement, PluginManager
+from damast.core.transformations import PipelineElement, PluginManager, plugin_manager
 
 
 class PluginsParser(BaseParser):
@@ -21,5 +21,26 @@ def execute(self, args):
                  f"{PluginManager.PLUGIN_PATH_ENV}={plugin_path})")
             return
 
-        for name, target in sorted(plugins.items()):
-            print(f"{name}: {target}")
+        # '<package>.<class>' -> 'module:class', grouped per plugin package
+        packages: dict[str, list[tuple[str, str]]] = {}
+        for qualified_name, target in plugins.items():
+            package, class_name = qualified_name.split(".", 1)
+            packages.setdefault(package, []).append((class_name, target.split(":")[0]))
+
+        for package, transformers in sorted(packages.items()):
+            source = self._describe_source(plugin_manager.resolve_requirement(transformers[0][1]))
+            print(f"{package}{f' ({source})' if source else ''}")
+
+            width = max(len(class_name) for class_name, _ in transformers)
+            for class_name, module_name in sorted(transformers):
+                # module relative to the package - omitted if defined in the package module itself
+                relative_module = module_name.removeprefix(package)
+                print(f"    {class_name:<{width}}  {relative_module}".rstrip())
+
+    @staticmethod
+    def _describe_source(requirement: dict[str, str] | None) -> str:
+        if not requirement:
+            return ""
+        if requirement.get("distribution"):
+            return f"{requirement['distribution']}=={requirement['version']}"
+        return f"local: {requirement['path']}"

src/damast/core/polars_dataframe.py

@@ -631,6 +631,8 @@ def read_batches(with_columns: list[str] | None,
                         n_rows -= df.height
                     yield df
 
+        # According to polars documentation this functionality is considered unstable
+        # https://docs.pola.rs/api/python/stable/reference/api/polars.io.plugins.register_io_source.html
         return register_io_source(read_batches, schema=schema), variables
 
     @staticmethod