enhancement/plugin manager

README.md

@@ -61,7 +61,7 @@ class JoinByTimestamp(PipelineElement):
         other_timestamp = self.get_name('timestamp', datasource='other')
         df_timestamp = self.get_name('timestamp')
 
-        df._dataframe = df.join(other._dataframe, left_on=df_timestamp, right_on=other_timestamp)
+        df.lazyframe = df.join(other.lazyframe, left_on=df_timestamp, right_on=other_timestamp)
         return df
 ```
 

docs/cli.rst

@@ -16,6 +16,9 @@ in sub-commands:
 **process**
     apply a data pipeline to a dataset
 
+**plugins**
+    list transformer plugins registered by installed packages or via ``DAMAST_PLUGIN_PATH``
+
 
 Inspect
 --------
@@ -237,3 +240,44 @@ Examples
 
 .. highlight:: none
 
+
+
+Plugins
+---------
+
+.. literalinclude:: ./examples/damast-plugins-help.txt
+  :language: none
+
+A pipeline can use transformers that are not part of the ``damast`` package itself. Two ways of
+registering such plugin transformers are supported - see :class:`damast.core.transformations.PluginManager`
+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``::
+
+      [project.entry-points."damast.transformers"]
+      MyTransformer = "acme_pkg.transformers:MyTransformer"
+
+- 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
+
+``damast plugins`` lists everything that is currently discoverable from either source, without
+requiring any Python code:
+
+.. highlight:: python
+
+::
+
+    export DAMAST_PLUGIN_PATH=/path/to/my/transformers
+    damast plugins
+
+    MyTripler: my_transformers:MyTripler
+
+.. highlight:: none
+
+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.
+

docs/usage.ipynb

@@ -227,7 +227,7 @@
     "        other_timestamp = self.get_name('timestamp', datasource='other')                                                          \n",
     "        df_timestamp = self.get_name('timestamp')                                                                                 \n",
     "                                                                                                                                  \n",
-    "        df._dataframe = df.join(other._dataframe, left_on=df_timestamp, right_on=other_timestamp)                                 \n",
+    "        df.lazyframe = df.join(other.lazyframe, left_on=df_timestamp, right_on=other_timestamp)                                 \n",
     "        return df\n",
     "\n",
     "event_time = data.dataframe.drop_nulls().select(polars.col('date_time_utc')).item(0,0)\n",
@@ -303,6 +303,116 @@
     "joined_adf.head(10).collect()"
    ]
   },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Plugin transformers\n",
+    "\n",
+    "Not every transformer needs to be part of the `damast` package - or even part of an installed\n",
+    "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.\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."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import os\n",
+    "import tempfile\n",
+    "from pathlib import Path\n",
+    "\n",
+    "plugin_dir = Path(tempfile.mkdtemp())\n",
+    "(plugin_dir / \"my_transformers.py\").write_text(\"\"\"\n",
+    "import polars\n",
+    "\n",
+    "from damast.core.dataframe import AnnotatedDataFrame\n",
+    "from damast.core.decorators import describe, input, output\n",
+    "from damast.core.transformations import PipelineElement\n",
+    "\n",
+    "\n",
+    "class 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",
+    "\n",
+    "os.environ[\"DAMAST_PLUGIN_PATH\"] = str(plugin_dir)\n",
+    "\n",
+    "from damast.core.transformations import PipelineElement\n",
+    "\n",
+    "# directories are only scanned once per process, so force a rescan after\n",
+    "# changing DAMAST_PLUGIN_PATH or adding/editing files\n",
+    "PipelineElement.reload_plugins()\n",
+    "PipelineElement.list_plugins()"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "`my_transformers.py` is now importable by its filename stem, and can be used in a pipeline like\n",
+    "any other transformer. Saving the pipeline records where the transformer came from, under\n",
+    "`requires`, so a pipeline loaded elsewhere fails with an actionable error (pointing at\n",
+    "`DAMAST_PLUGIN_PATH`) instead of a bare import error if the local file is missing."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import my_transformers  # the module we just wrote to `plugin_dir`\n",
+    "from damast.core.dataprocessing import DataProcessingPipeline\n",
+    "\n",
+    "\n",
+    "plugin_pipeline = DataProcessingPipeline(name=\"plugin-example\", base_dir=\"./output_dir\") \\\n",
+    "    .add(\"double_mmsi\", my_transformers.Doubler(), name_mappings={\"x\": \"mmsi\"})\n",
+    "\n",
+    "pipeline_path = plugin_pipeline.save(\"./output_dir\")\n",
+    "print(pipeline_path.read_text())"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "### Installable plugin packages (entry-points)\n",
+    "\n",
+    "For a transformer that should be pip-installable and reused across projects, a package can\n",
+    "advertise its `PipelineElement` subclasses via the `damast.transformers` entry-point group in its\n",
+    "own `pyproject.toml`:\n",
+    "\n",
+    "```toml\n",
+    "[project.entry-points.\"damast.transformers\"]\n",
+    "MyTransformer = \"acme_pkg.transformers:MyTransformer\"\n",
+    "```\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."
+   ]
+  },
   {
    "cell_type": "code",
    "execution_count": null,

pyproject.toml

@@ -65,7 +65,11 @@ homepage = "https://simula.github.io/damast"
 documentation = "https://simula.github.io/damast"
 repository = "https://github.com/simula/damast"
 
-#[project.entry-points]
+# 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"
+# Discoverable via `damast plugins` or PipelineElement.list_plugins().
 
 [project.optional-dependencies]
 dev = [

src/damast/cli/data_annotate.py

@@ -190,7 +190,7 @@ def update(self, args, files):
                 if column_spec.representation_type:
                     representation_type = adf.set_dtype(column_spec.name, column_spec.representation_type)
                     metadata[column_spec.name].representation_type = representation_type
-                    metadata[column_spec.name].update_datarange_and_stats(adf._dataframe, column_spec.name)
+                    metadata[column_spec.name].update_datarange_and_stats(adf.lazyframe, column_spec.name)
 
         adf._metadata = metadata
         adf.validate_metadata(ValidationMode.UPDATE_DATA)

src/damast/cli/data_inspect.py

@@ -33,11 +33,14 @@ def __init__(self, parser: ArgumentParser):
 
         parser.add_argument("--filter",
                 help="Filter based on column data, e.g., mmsi==120123, allowed operators !=,<,>,==,<=,>=,<> or =~ for a regex, e.g.,'name =~ '^SAR*'",
-                action="append"
+                nargs="+",
+                type=str,
+                required=False
         )
         parser.add_argument("--head", type=int, default=10, help="First this number of rows, default is 10")
         parser.add_argument("--tail", type=int, default=10, help="Print number of rows from the end, default is 10")
         parser.add_argument("--column-count", type=int, default=10, help="Number of columns to show")
+        parser.add_argument("--column-width", type=int, default=None, help="Column width to show")
 
         parser.add_argument("--columns",
                 help="Show/Select these columns",
@@ -114,18 +117,18 @@ def execute(self, args):
                         else:
                             logger.warning(f"Filter expression invalid: {filter_expression}")
 
-                    adf._dataframe = eval(f"adf._dataframe{filter_values}")
-                    adf._metadata = AnnotatedDataFrame.infer_annotation(adf._dataframe)
+                    adf.lazyframe = eval(f"adf.lazyframe{filter_values}")
+                    adf._metadata = AnnotatedDataFrame.infer_annotation(adf.lazyframe)
 
                 print(adf.metadata.to_str(columns=args.columns))
                 print(f"\n\nFirst {args.head} and last {args.tail} rows:")
-                df = adf._dataframe
+                df = adf.lazyframe
                 if args.columns:
                     df = df.select(args.columns)
 
-                with pl.Config(tbl_rows=args.head, tbl_cols=args.column_count):
+                with pl.Config(tbl_rows=args.head, tbl_cols=args.column_count, fmt_str_lengths=args.column_width):
                     print(df.head(n=args.head).collect())
-                with pl.Config(tbl_rows=args.tail, tbl_cols=args.column_count):
+                with pl.Config(tbl_rows=args.tail, tbl_cols=args.column_count, fmt_str_lengths=args.column_width):
                     print(df.tail(n=args.tail).collect())
 
         except RuntimeError as e:

src/damast/cli/main.py

@@ -14,6 +14,7 @@
 from damast.cli.data_inspect import DataInspectParser
 from damast.cli.data_processing import DataProcessingParser
 from damast.cli.experiment import ExperimentParser
+from damast.cli.plugins import PluginsParser
 from damast.config import DAMAST_LOG_DATE_FORMAT, DAMAST_LOG_FORMAT, DAMAST_LOG_STYLE
 
 logging.basicConfig(
@@ -87,6 +88,10 @@ def run():
                                          help="Process data by running a predefined pipeline",
                                          parser_klass=DataProcessingParser)
 
+    main_parser.attach_subcommand_parser(subcommand="plugins",
+                                         help="List transformer plugins registered by installed packages",
+                                         parser_klass=PluginsParser)
+
 
     args = main_parser.parse_args()
     if args.version:

src/damast/cli/plugins.py

@@ -0,0 +1,25 @@
+import os
+from argparse import ArgumentParser
+
+from damast.cli.base import BaseParser
+from damast.core.transformations import PipelineElement, PluginManager
+
+
+class PluginsParser(BaseParser):
+    def __init__(self, parser: ArgumentParser):
+        super().__init__(parser=parser)
+
+        parser.description = ("damast plugins - list transformer plugins registered by installed packages"
+                              f" (entry-point group '{PluginManager.ENTRY_POINT_GROUP}') or via the"
+                              f" '{PluginManager.PLUGIN_PATH_ENV}' environment variable")
+
+    def execute(self, args):
+        plugins = PipelineElement.list_plugins()
+        if not plugins:
+            plugin_path = os.environ.get(PluginManager.PLUGIN_PATH_ENV, "<unset>")
+            print(f"No transformer plugins registered (entry-point group '{PluginManager.ENTRY_POINT_GROUP}', "
+                 f"{PluginManager.PLUGIN_PATH_ENV}={plugin_path})")
+            return
+
+        for name, target in sorted(plugins.items()):
+            print(f"{name}: {target}")

src/damast/core/dataframe.py

@@ -49,17 +49,14 @@ class AnnotatedDataFrame(XDataFrame):
     #: Metadata associated with the dataframe
     _metadata: MetaData
 
-    #: The actual dataframe
-    _dataframe: DataFrame
-
     def __init__(
         self,
         dataframe: polars.DataFrame | polars.LazyFrame | XDataFrame,
         metadata: MetaData,
         validation_mode: ValidationMode = ValidationMode.READONLY,
     ):
         if isinstance(dataframe, XDataFrame):
-            dataframe = dataframe._dataframe
+            dataframe = dataframe.lazyframe
 
         if isinstance(dataframe, polars.DataFrame):
             dataframe = dataframe.lazy()
@@ -93,8 +90,8 @@ def ensure_type(cls, obj: any):
         if not isinstance(obj, cls):
             raise ValueError("Object {obj} is not an AnnotatedDataFrame")
 
-        if not isinstance(obj._dataframe, DataFrame):
-            raise ValueError(f"AnnotatedDataFrame._dataframe is not of type {DataFrame}")
+        if not isinstance(obj.lazyframe, DataFrame):
+            raise ValueError(f"AnnotatedDataFrame.lazyframe is not of type {DataFrame}")
 
     def validate_metadata(
         self, validation_mode: ValidationMode = ValidationMode.READONLY
@@ -106,15 +103,15 @@ def validate_metadata(
         :raise RuntimeError: Dependending on the validation mode an exception will be raise to ensure the data spec
                conformance
         """
-        self._dataframe = self._metadata.apply(df=self._dataframe, validation_mode=validation_mode)
+        self.lazyframe = self._metadata.apply(df=self.lazyframe, validation_mode=validation_mode)
 
     def is_empty(self) -> bool:
         """
         Check if annotated dataframe has associated data.
 
-        :return: False if there is an internal :code:`_dataframe` set, True otherwise
+        :return: False if there is an internal :code:`lazyframe` set, True otherwise
         """
-        return self._dataframe is None
+        return self.lazyframe is None
 
     def get_fulfillment(
         self, expected_specs: list[DataSpecification]
@@ -169,7 +166,7 @@ def save(self, *, filename: Union[str, Path]) -> AnnotatedDataFrame:
         :param filename: Filename to use for saving
 
         """
-        if self._dataframe is None:
+        if self.lazyframe is None:
             raise ValueError(f"{self.__class__.__name__}.save: no dataframe to save")
 
         metadata_filename = Path(filename).with_suffix(DAMAST_SPEC_SUFFIX)
@@ -180,7 +177,7 @@ def save(self, *, filename: Union[str, Path]) -> AnnotatedDataFrame:
             return self
 
         # First save the hdf5 file in order to add then the metadata to it
-        XDataFrame.export_hdf5(self._dataframe, filename)
+        XDataFrame.export_hdf5(self.lazyframe, filename)
         self._metadata.append_to_hdf(filename)
 
         return self
@@ -189,7 +186,7 @@ def export(self, filename: str | Path):
         """
         Export the annotated dataframe to a file. By default the format is parquet.
         """
-        arrow_table = self._dataframe.compat.collected().to_arrow()
+        arrow_table = self.lazyframe.compat.collected().to_arrow()
         new_schema = arrow_table.schema.with_metadata({b'annotated_dataframe': json.dumps(dict(self._metadata), default=str).encode('UTF-8')})
         arrow_table = pyarrow.Table.from_arrays(arrow_table.columns, schema=new_schema)
         pq.write_table(arrow_table, filename)
@@ -442,7 +439,7 @@ def convert_csv_to_adf(
         adf.save(filename=output_filename)
 
     def drop(self, columns, strict: bool = True) -> AnnotatedDataFrame:
-        self._dataframe = self._dataframe.drop(columns, strict=strict)
+        self.lazyframe = self.lazyframe.drop(columns, strict=strict)
         self._metadata = self._metadata.drop(columns)
         return self
 
@@ -451,8 +448,8 @@ def copy(self):
 
     @property
     def shape(self):
-        return self._dataframe.compat.collected().shape
+        return self.lazyframe.compat.collected().shape
 
     def __deepcopy__(self, memo=None):
         # ignore validation, also erroneous frames should be copyable
-        return AnnotatedDataFrame(self._dataframe.clone(), copy.deepcopy(self._metadata), validation_mode=ValidationMode.IGNORE)
+        return AnnotatedDataFrame(self.lazyframe.clone(), copy.deepcopy(self._metadata), validation_mode=ValidationMode.IGNORE)