diff --git a/README.md b/README.md index 331ca4d..72614d1 100644 --- a/README.md +++ b/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 ``` diff --git a/docs/cli.rst b/docs/cli.rst index bf01640..bd19438 100644 --- a/docs/cli.rst +++ b/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. + diff --git a/docs/usage.ipynb b/docs/usage.ipynb index ad8d785..48638bb 100644 --- a/docs/usage.ipynb +++ b/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, diff --git a/pyproject.toml b/pyproject.toml index c6e608b..7b9572d 100644 --- a/pyproject.toml +++ b/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 = [ diff --git a/src/damast/cli/data_annotate.py b/src/damast/cli/data_annotate.py index c6334ee..f733d59 100644 --- a/src/damast/cli/data_annotate.py +++ b/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) diff --git a/src/damast/cli/data_inspect.py b/src/damast/cli/data_inspect.py index 2cfb5e9..3666f7f 100644 --- a/src/damast/cli/data_inspect.py +++ b/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: diff --git a/src/damast/cli/main.py b/src/damast/cli/main.py index 7b508bb..4cc3d50 100644 --- a/src/damast/cli/main.py +++ b/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: diff --git a/src/damast/cli/plugins.py b/src/damast/cli/plugins.py new file mode 100644 index 0000000..23fc83b --- /dev/null +++ b/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, "") + 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}") diff --git a/src/damast/core/dataframe.py b/src/damast/core/dataframe.py index 3e52bd8..bd0e127 100644 --- a/src/damast/core/dataframe.py +++ b/src/damast/core/dataframe.py @@ -49,9 +49,6 @@ class AnnotatedDataFrame(XDataFrame): #: Metadata associated with the dataframe _metadata: MetaData - #: The actual dataframe - _dataframe: DataFrame - def __init__( self, dataframe: polars.DataFrame | polars.LazyFrame | XDataFrame, @@ -59,7 +56,7 @@ def __init__( 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) diff --git a/src/damast/core/dataprocessing.py b/src/damast/core/dataprocessing.py index 459734d..4b22b32 100644 --- a/src/damast/core/dataprocessing.py +++ b/src/damast/core/dataprocessing.py @@ -299,8 +299,8 @@ def save_state(self, :param dir: directory where to save this pipeline """ filename = Path(dir) / f"{self.name}{VAEX_STATE_SUFFIX}" - #df._dataframe.state_write(file=filename) - df._dataframe.serialize(filename) + #df.lazyframe.state_write(file=filename) + df.lazyframe.serialize(filename) return filename @classmethod @@ -369,7 +369,7 @@ def load_state( ) #df.dataframe.state_load(file=filename) - df._dataframe = df.dataframe.deserialize(filename) + df.lazyframe = df.dataframe.deserialize(filename) return df def prepare(self, diff --git a/src/damast/core/metadata.py b/src/damast/core/metadata.py index c68f9b2..e8e4ef9 100644 --- a/src/damast/core/metadata.py +++ b/src/damast/core/metadata.py @@ -567,7 +567,7 @@ def apply( elif df_type == pl.DataFrame: df = df.lazy() elif df_type == XDataFrame: - df = df._dataframe + df = df.lazyframe else: raise TypeError("MetaData.apply: dataframe must be either " " polars.LazyFrame," @@ -622,12 +622,12 @@ def apply( logger.info( f"Filtering out for column '{column_name}' values that are out of range: {self.value_range}." ) - xdf._dataframe = xdf._dataframe.filter( + xdf.lazyframe = xdf.lazyframe.filter( (pl.col(column_name) >= self.value_range.min) & (pl.col(column_name) <= self.value_range.max) ) else: - xdf._dataframe = xdf._dataframe.with_columns( + xdf.lazyframe = xdf.lazyframe.with_columns( pl.when( (pl.col(column_name) < self.value_range.min) | (pl.col(column_name) > self.value_range.max) @@ -635,7 +635,7 @@ def apply( .otherwise(pl.col(column_name)) .alias(column_name) ) - return xdf._dataframe + return xdf.lazyframe if validation_mode == ValidationMode.UPDATE_METADATA: if self.representation_type is None: @@ -667,7 +667,7 @@ def update_datarange_and_stats(self, df, column_name: str): except ValueError as e: logger.debug(f"Metadata.update_datarange_and_stats: could not update datarange and stats '{column_name}' -- {e}") - return xdf._dataframe + return xdf.lazyframe def get_fulfillment(self, data_spec: DataSpecification) -> Fulfillment: """ diff --git a/src/damast/core/polars_dataframe.py b/src/damast/core/polars_dataframe.py index 04a668e..be31123 100644 --- a/src/damast/core/polars_dataframe.py +++ b/src/damast/core/polars_dataframe.py @@ -1,6 +1,7 @@ from __future__ import annotations import logging +import os import re from pathlib import Path from typing import ClassVar @@ -21,7 +22,16 @@ VAEX_HDF5_ROOT: str = "/table" VAEX_HDF5_COLUMNS: str = f"{VAEX_HDF5_ROOT}/columns" -polars.Config.set_engine_affinity("streaming") +# Prefer an engine affinity the embedding application/environment has already configured +# (via POLARS_ENGINE_AFFINITY or a prior polars.Config.set_engine_affinity call) over +# unconditionally overriding it - only apply damast's own default when nothing is set yet. +if "POLARS_ENGINE_AFFINITY" in os.environ: + logger.warning( + "damast.core.polars_dataframe: POLARS_ENGINE_AFFINITY is already set to" + f" '{os.environ['POLARS_ENGINE_AFFINITY']}' - keeping it instead of damast's default 'streaming'" + ) +else: + polars.Config.set_engine_affinity("streaming") POLARS_TYPE_DICT = { key: value @@ -42,16 +52,30 @@ def __getattr__(cls, attr_name): @polars.api.register_dataframe_namespace("compat") @polars.api.register_lazyframe_namespace("compat") class PolarsDataFrame(metaclass=Meta): - _dataframe: LazyFrame _polars_dataframe: PolarsDataFrame _dataframe_collected: polars.DataFrame def __init__(self, df: LazyFrame | polars.DataFrame): + self.lazyframe = df + + @property + def lazyframe(self) -> LazyFrame: + """ + The underlying ``polars.LazyFrame``. + + This is the sole point of mutation for the wrapped dataframe - assigning to it (rather + than e.g. a plain, private instance attribute) is what lets us keep the ``collected()`` + cache and the ``dataframe`` accessor consistent with the data that is actually stored, + instead of silently returning a stale snapshot after an update. + """ + return self.__lazyframe + + @lazyframe.setter + def lazyframe(self, df: LazyFrame | polars.DataFrame): if type(df) is polars.DataFrame: - self._dataframe = df.lazy() - else: - self._dataframe = df + df = df.lazy() + self.__lazyframe = df self._dataframe_collected = None self._polars_dataframe = None @@ -70,7 +94,10 @@ def resolve_type(cls, type_txt: str): elif type_txt == "float": type_txt = "Float64" - return eval(type_txt, cls.types()) + try: + return cls.types()[type_txt] + except KeyError: + raise TypeError(f"{cls.__name__}.resolve_type: unknown polars type '{type_txt}'") @property @@ -84,14 +111,14 @@ def dataframe(self) -> PolarsDataFrame: :return: The underlying dataframe """ - if self._polars_dataframe is None or self._dataframe is not self._polars_dataframe._dataframe: - self._polars_dataframe = PolarsDataFrame(self._dataframe) + if self._polars_dataframe is None or self.lazyframe is not self._polars_dataframe.lazyframe: + self._polars_dataframe = PolarsDataFrame(self.lazyframe) return self._polars_dataframe def collected(self): if self._dataframe_collected is None: - self._dataframe_collected = self._dataframe.collect() + self._dataframe_collected = self.lazyframe.collect() return self._dataframe_collected @@ -114,7 +141,7 @@ def __getitem__(self, column_name: str): :param item: Name of the key when using [] operators :return: item/column from the underlying vaex.dataframe """ - return self._dataframe.select(column_name) + return self.lazyframe.select(column_name) def ensure_column(self, column_name: str): """ @@ -128,14 +155,14 @@ def column_names(self) -> list[str]: """ Get all column names (without collecting the full dataframe) """ - return self._dataframe.collect_schema().names() + return self.lazyframe.collect_schema().names() def dtype(self, column_name: str) -> polars.datatypes.DataType: """ Get column dtype (without collecting the full dataframe) """ idx = self.column_names.index(column_name) - return self._dataframe.collect_schema().dtypes()[idx] + return self.lazyframe.collect_schema().dtypes()[idx] def set_dtype(self, column_name, representation_type) -> polars.datatype.DataType: """ @@ -151,7 +178,7 @@ def set_dtype(self, column_name, representation_type) -> polars.datatype.DataTyp if hasattr(polars, representation_type): representation_type = getattr(polars, representation_type) - self._dataframe = self._dataframe.with_columns(polars.col(column_name).cast(representation_type).alias(column_name)) + self.lazyframe = self.lazyframe.with_columns(polars.col(column_name).cast(representation_type).alias(column_name)) return representation_type def minmax(self, column_name: str) -> tuple[any, any]: @@ -161,7 +188,7 @@ def minmax(self, column_name: str) -> tuple[any, any]: self.ensure_column(column_name) try: - result = self._dataframe.select([ + result = self.lazyframe.select([ polars.col(column_name).min().alias("min_value"), polars.col(column_name).max().alias("max_value") ]).collect() @@ -177,7 +204,7 @@ def categories(self, column_name: str, max_count: int = 100) -> list[str]: self.ensure_column(column_name) try: - categories = self._dataframe.select(column_name).unique().sort(by=column_name).collect()[:,0].to_list() + categories = self.lazyframe.select(column_name).unique().sort(by=column_name).collect()[:,0].to_list() except Exception as e: raise RuntimeError(f"Failed to extract categories for column '{column_name}' -- {e}") from e @@ -224,7 +251,7 @@ def minmax_stats(self, column_names: list[str]) -> dict[str, dict[str, any]]: polars.col(column).std().alias(f"{column}_stddev"), ]) - result = self._dataframe.select( + result = self.lazyframe.select( fields ).collect() @@ -256,7 +283,7 @@ def minmax_stats(self, column_names: list[str]) -> dict[str, dict[str, any]]: def stats(self, column_name: str) -> NumericValueStats: self.ensure_column(column_name) - result = self._dataframe.select([ + result = self.lazyframe.select([ polars.col(column_name).mean().alias("mean"), polars.col(column_name).std().alias("stddev"), polars.col(column_name).count().alias("total_count"), @@ -279,13 +306,13 @@ def __getattr__(self, attr_name): """ # allow dataframe.col_one if attr_name in self.column_names: - return self._dataframe.select(attr_name) + return self.lazyframe.select(attr_name) if attr_name in ["__setstate__", "__getstate__"]: raise AttributeError(f"{self.__class__.__name__}.__getattr__: {attr_name} does not exist") """ Called for failed attribute accesses so forwarding to underlying polars frame """ - return getattr(self._dataframe, attr_name) + return getattr(self.lazyframe, attr_name) def __setitem__(self, key, values): """ @@ -297,7 +324,7 @@ def __setitem__(self, key, values): if type(values) is polars.LazyFrame: values = values.collect().to_numpy() - self._dataframe = self._dataframe.with_columns( + self.lazyframe = self.lazyframe.with_columns( polars.Series( name=key, values=values diff --git a/src/damast/core/transformations.py b/src/damast/core/transformations.py index db2d94a..fe1dcb5 100644 --- a/src/damast/core/transformations.py +++ b/src/damast/core/transformations.py @@ -2,9 +2,16 @@ import copy import importlib +import importlib.metadata +import importlib.util import inspect +import os import re +import sys from abc import abstractmethod +from logging import getLogger +from pathlib import Path +from types import ModuleType import numpy as np import polars @@ -20,6 +27,188 @@ ) from .formatting import DEFAULT_INDENT +logger = getLogger(__name__) + + +class PluginManager: + """ + Discovers and resolves :class:`PipelineElement` 'plugin' transformers, i.e. + transformers that are not necessarily part of the damast package itself. + + Two plugin sources are supported: + + - packages that register :class:`PipelineElement` subclasses via the + ``damast.transformers`` entry-point group, e.g. in their own pyproject.toml:: + + [project.entry-points."damast.transformers"] + MyTransformer = "acme_pkg.transformers:MyTransformer" + + - loose ``*.py`` files in directories listed in the ``DAMAST_PLUGIN_PATH`` + environment variable (os.pathsep-separated), for local/ad-hoc transformers that + are not part of an installed package. Every top-level file found there is + imported once (using its filename stem as 'module_name'), so that any + :class:`PipelineElement` subclasses it defines become resolvable exactly like + classes from an installed package. + """ + + #: Entry-point group that plugin packages use to advertise PipelineElement subclasses + ENTRY_POINT_GROUP = "damast.transformers" + + #: Environment variable with an os.pathsep-separated list of local plugin directories + PLUGIN_PATH_ENV = "DAMAST_PLUGIN_PATH" + + def __init__(self): + #: module_name -> loaded module, for modules imported from PLUGIN_PATH_ENV + self._local_modules: dict[str, ModuleType] = {} + #: module_name -> source file, used to detect/warn about name collisions + self._local_files: dict[str, Path] = {} + self._loaded = False + self._requirement_cache: dict[str, dict[str, str] | None] = {} + + @property + def local_modules(self) -> dict[str, ModuleType]: + return dict(self._local_modules) + + @property + def local_files(self) -> dict[str, Path]: + return dict(self._local_files) + + def plugin_path_dirs(self) -> list[Path]: + raw = os.environ.get(self.PLUGIN_PATH_ENV, "") + return [Path(p) for p in raw.split(os.pathsep) if p.strip()] + + def load_local_plugins(self, force: bool = False) -> dict[str, ModuleType]: + """ + Import loose '*.py' files found in :attr:`PLUGIN_PATH_ENV` directories, so that + any PipelineElement subclasses they define become resolvable by + 'module_name'/'class_name' - the same way as classes from an installed package. + + :param force: Re-scan the configured directories and re-import their files, even + if they were already loaded in this process + """ + if self._loaded and not force: + return self._local_modules + + if force: + self._local_modules.clear() + self._local_files.clear() + self._requirement_cache.clear() + + for plugin_dir in self.plugin_path_dirs(): + if not plugin_dir.is_dir(): + logger.warning(f"PluginManager: {self.PLUGIN_PATH_ENV} entry '{plugin_dir}'" + " is not a directory - skipping") + continue + + for py_file in sorted(plugin_dir.glob("*.py")): + module_name = py_file.stem + if module_name.startswith("_"): + continue + + existing_file = self._local_files.get(module_name) + if existing_file is not None: + if existing_file != py_file: + logger.warning( + f"PluginManager: plugin module '{module_name}' from '{py_file}' collides with" + f" already loaded '{existing_file}' - keeping the first one" + ) + continue + + spec = importlib.util.spec_from_file_location(module_name, py_file) + module = importlib.util.module_from_spec(spec) + try: + spec.loader.exec_module(module) + except Exception as e: + logger.warning(f"PluginManager: failed to load plugin '{py_file}': {e}") + continue + + sys.modules[module_name] = module + self._local_modules[module_name] = module + self._local_files[module_name] = py_file + + self._loaded = True + return self._local_modules + + def reload(self): + """ + Force re-scanning of :attr:`PLUGIN_PATH_ENV` directories and re-importing their files. + + Useful after the environment variable was changed, or files were added/edited, since + directories are otherwise only scanned once per process. + """ + self.load_local_plugins(force=True) + + def resolve_requirement(self, module_name: str) -> dict[str, str] | None: + """ + Identify what installable distribution package - or local plugin file - provides + ``module_name``. + + This is used to record which package (or local file) a transformer originates from + when a pipeline is saved, so that loading the pipeline elsewhere can point users at + the missing package/file instead of failing with a bare :class:`ImportError`. + + :param module_name: Dotted module path of a :class:`PipelineElement` subclass + :return: Dict with 'distribution' and 'version' for an installed package; a dict with + 'hint': 'local' and 'path' for a transformer loaded from :attr:`PLUGIN_PATH_ENV`; + or None if it could not be resolved at all (e.g. the class is defined in a script or + notebook that is neither installed nor on the plugin path) + """ + if module_name in self._requirement_cache: + return self._requirement_cache[module_name] + + self.load_local_plugins() + + result = None + local_file = self._local_files.get(module_name) + if local_file is not None: + result = {"hint": "local", "path": str(local_file)} + else: + top_level = module_name.split(".")[0] + try: + distributions = importlib.metadata.packages_distributions().get(top_level) + except Exception: + distributions = None + + if distributions: + distribution = distributions[0] + try: + version = importlib.metadata.version(distribution) + result = {"distribution": distribution, "version": version} + except importlib.metadata.PackageNotFoundError: + result = None + + self._requirement_cache[module_name] = result + return result + + def list_plugins(self) -> dict[str, str]: + """ + Discover transformer plugins from both the entry-point group and local plugin path. + + This is purely a discovery/documentation aid - :func:`PipelineElement.create_new` + resolves classes by ``module_name``/``class_name`` regardless of whether they are + registered here. + + :return: Mapping of class name to its 'module_name:class_name' target + """ + plugins = { + ep.name: ep.value + for ep in importlib.metadata.entry_points(group=self.ENTRY_POINT_GROUP) + } + + for module_name, module in self.load_local_plugins().items(): + for attr_name, obj in vars(module).items(): + if (inspect.isclass(obj) + and issubclass(obj, PipelineElement) + and obj is not PipelineElement + and obj.__module__ == module_name): + plugins[attr_name] = f"{module_name}:{attr_name}" + + return plugins + + +#: Default, process-wide plugin manager used by PipelineElement +plugin_manager = PluginManager() + class Transformer: uuid: str @@ -190,20 +379,76 @@ def output_specs(self) -> list[DataSpecification]: return specs + @classmethod + def _missing_plugin_message(cls, + module_name: str, + class_name: str, + requires: dict[str, str] | None) -> str: + if requires and requires.get("distribution"): + pip_spec = requires["distribution"] + if requires.get("version"): + pip_spec += f"=={requires['version']}" + return (f"{cls.__name__}.create_new: could not load transformer '{class_name}' from '{module_name}'. " + f"This pipeline requires the plugin package '{pip_spec}', which is not installed. " + f"Install it with: pip install {pip_spec}") + + plugin_path = os.environ.get(PluginManager.PLUGIN_PATH_ENV, "") + origin_hint = "" + if requires and requires.get("hint") == "local": + origin_hint = f" It was saved as a local transformer, originally loaded from '{requires.get('path')}'." + + return (f"{cls.__name__}.create_new: could not load '{class_name}' from '{module_name}'.{origin_hint} " + "Ensure that the package providing this transformer is installed and importable, " + f"or - if it is a local/ad-hoc transformer - that the directory containing " + f"'{module_name}.py' is listed in the '{PluginManager.PLUGIN_PATH_ENV}' environment variable" + f" (currently: {plugin_path}).") + + @classmethod + def _check_requirement(cls, + module_name: str, + class_name: str, + requires: dict[str, str]): + distribution = requires.get("distribution") + if not distribution: + return + + try: + installed_version = importlib.metadata.version(distribution) + except importlib.metadata.PackageNotFoundError: + raise ImportError(cls._missing_plugin_message(module_name, class_name, requires)) + + expected_version = requires.get("version") + if expected_version and installed_version != expected_version: + logger.warning( + f"{cls.__name__}.create_new: '{class_name}' was saved with plugin package" + f" '{distribution}=={expected_version}', but '{installed_version}' is installed." + " Results may differ from when the pipeline was created." + ) + @classmethod def create_new(cls, module_name: str, class_name: str, name_mappings: dict[str, dict[str, str]] | None = None, - parameters: dict[str, any] | None = {}) -> PipelineElement: + parameters: dict[str, any] | None = {}, + requires: dict[str, str] | None = None) -> PipelineElement: """ Create a new PipelineElement Subclass instance dynamically :param module_name: Name of the module for the PipelineElement class :param class_name: Name of the PipelineElement subclass :param name_mappings: Dictionary of name mappings that should apply + :param requires: Optional info on where this transformer was resolved from when the + pipeline was saved - see :func:`__iter__`. Either an installed plugin package + ('distribution' + 'version'), or a local plugin file ('hint': 'local' + 'path'). + Used to give an actionable error when the providing package/file is missing. :return: Instance for the PipelineElement instance + .. note:: + If the class cannot be found on the regular import path, directories listed in + the ``DAMAST_PLUGIN_PATH`` environment variable are scanned for a matching + '.py' file before giving up - see :func:`list_plugins`. + :raise ValueError: If module or class with given name is not specified :raise ImportError: If class could not be loaded """ @@ -213,11 +458,20 @@ def create_new(cls, if class_name is None: raise ValueError(f"{cls.__name__}.create_new: missing 'class_name'") - p_module = importlib.import_module(module_name) + if requires: + cls._check_requirement(module_name=module_name, class_name=class_name, requires=requires) + + plugin_manager.load_local_plugins() + + try: + p_module = importlib.import_module(module_name) + except ImportError: + raise ImportError(cls._missing_plugin_message(module_name, class_name, requires)) + if hasattr(p_module, class_name): klass = getattr(p_module, class_name) else: - raise ImportError(f"{cls.__name__}.create_new: could not load '{class_name}' from '{p_module}'") + raise ImportError(cls._missing_plugin_message(module_name, class_name, requires)) if parameters: instance = klass(**parameters) else: @@ -229,11 +483,31 @@ def create_new(cls, instance._name_mappings = name_mappings return instance + @classmethod + def list_plugins(cls) -> dict[str, str]: + """ + Discover transformer plugins - see :class:`PluginManager` for details on the two + supported sources (installed packages via entry-points, and local files via + ``DAMAST_PLUGIN_PATH``). + + :return: Mapping of class name to its 'module_name:class_name' target + """ + return plugin_manager.list_plugins() + + @classmethod + def reload_plugins(cls): + """ + Force re-scanning of ``DAMAST_PLUGIN_PATH`` directories and re-importing their files - + see :func:`PluginManager.reload`. + """ + plugin_manager.reload() + def __iter__(self): yield "module_name", f"{self.__class__.__module__}" yield "class_name", f"{self.__class__.__qualname__}" yield "parameters", self.parameters yield "name_mappings", self.name_mappings + yield "requires", plugin_manager.resolve_requirement(self.__class__.__module__) def __eq__(self, other): return dict(self) == dict(other) @@ -316,10 +590,10 @@ def transform(self, df: AnnotatedDataFrame): clone = df.copy() for feature in self.features: - clone._dataframe = clone._dataframe.with_columns( + clone.lazyframe = clone.lazyframe.with_columns( (np.cos(polars.col(feature)*2*np.pi) / self.n).alias(f"{feature}_x") ) - clone._dataframe = clone._dataframe.with_columns( + clone.lazyframe = clone.lazyframe.with_columns( (np.cos(polars.col(feature)*2*np.pi) / self.n).alias(f"{feature}_y") ) return clone diff --git a/src/damast/data_handling/transformers/augmenters.py b/src/damast/data_handling/transformers/augmenters.py index 7db306d..fe39371 100644 --- a/src/damast/data_handling/transformers/augmenters.py +++ b/src/damast/data_handling/transformers/augmenters.py @@ -118,7 +118,7 @@ def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame: pl.col(self.dataset_column) ).lazy() - dataframe = df._dataframe.join( + dataframe = df.lazyframe.join( other=other_df, left_on=self.get_name("x"), right_on=self.right_on, @@ -127,7 +127,7 @@ def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame: how=self.join_how.value ) - df._dataframe = dataframe.rename({self.dataset_column: self.get_name("out")}) + df.lazyframe = dataframe.rename({self.dataset_column: self.get_name("out")}) return df @@ -207,7 +207,7 @@ def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame: Fill in values for NA and missing entries """ mapped_name = self.get_name("x") - df._dataframe = df._dataframe.with_columns( + df.lazyframe = df.lazyframe.with_columns( pl.col(mapped_name).fill_null(self.fill_value).alias(mapped_name), ).with_columns( pl.col(mapped_name).fill_nan(self.fill_value).alias(mapped_name) @@ -228,12 +228,12 @@ class AddLocalIndex(PipelineElement): @damast.core.output({"local_index": {"representation_type": int}, "reverse_{{local_index}}": {"representation_type": int}}) def transform(self, df: damast.core.AnnotatedDataFrame) -> damast.core.AnnotatedDataFrame: - dataframe = df._dataframe + dataframe = df.lazyframe group_column = self.get_name("group") sort_column = self.get_name("sort") - df._dataframe = dataframe\ + df.lazyframe = dataframe\ .sort(group_column, sort_column)\ .with_columns( pl.int_range(pl.len()).over(group_column).alias(self.get_name("local_index")), @@ -253,7 +253,7 @@ class AddDeltaTime(PipelineElement): "time_column": {}}) @damast.core.output({"delta_time": {"representation_type": float, "unit": "s"}}) def transform(self, df: damast.core.AnnotatedDataFrame) -> damast.core.AnnotatedDataFrame: - dataframe = df._dataframe + dataframe = df.lazyframe group_column = self.get_name("group") time_column = self.get_name("time_column") @@ -277,7 +277,7 @@ def transform(self, df: damast.core.AnnotatedDataFrame) -> damast.core.Annotated pl.col('delta_time').fill_null(0).alias('delta_time') ) - df._dataframe = dataframe + df.lazyframe = dataframe return df @@ -313,7 +313,7 @@ def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame: from_mapped_name = self.get_name("from") to_mapped_name = self.get_name("to") - df._dataframe = df._dataframe.with_columns( + df.lazyframe = df.lazyframe.with_columns( (pl.col(from_mapped_name).str.to_datetime().dt.timestamp("ms")/1000.0).alias(to_mapped_name) ) return df @@ -343,7 +343,7 @@ def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame: Multiply a column by a given value """ mapped_name = self.get_name("x") - df._dataframe = df._dataframe.with_columns( + df.lazyframe = df.lazyframe.with_columns( (pl.col(mapped_name)*self.mul_value).alias(mapped_name) ) return df @@ -381,7 +381,7 @@ def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame: input_mapped_name = self.get_name("x") output_mapped_name = self.get_name("y") - df._dataframe = df._dataframe.with_columns( + df.lazyframe = df.lazyframe.with_columns( pl.col(input_mapped_name).cast(self.new_type).alias(output_mapped_name) ) return df diff --git a/src/damast/data_handling/transformers/cycle_transformer.py b/src/damast/data_handling/transformers/cycle_transformer.py index cf4bf97..e49d43a 100644 --- a/src/damast/data_handling/transformers/cycle_transformer.py +++ b/src/damast/data_handling/transformers/cycle_transformer.py @@ -19,7 +19,7 @@ def __init__(self, n: int): }) def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame: feature = self.get_name('x') - df._dataframe = df._dataframe.with_columns( + df.lazyframe = df.lazyframe.with_columns( (np.sin(polars.col(feature)*2*np.pi) / self.n).alias(f"{feature}_x"), (np.cos(polars.col(feature)*2*np.pi) / self.n).alias(f"{feature}_y") ) @@ -41,7 +41,7 @@ class TimestampCycleTransformer(PipelineElement): def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame: feature = self.get_name('x') - df._dataframe = df._dataframe.with_columns( + df.lazyframe = df.lazyframe.with_columns( (np.sin(polars.col(feature).dt.quarter()*2*np.pi) / 4).alias(f"{feature}_quarter_x"), (np.cos(polars.col(feature).dt.quarter()*2*np.pi) / 4).alias(f"{feature}_quarter_y"), (np.sin(polars.col(feature).dt.week()*2*np.pi) / 53).alias(f"{feature}_week_x"), diff --git a/src/damast/data_handling/transformers/filters.py b/src/damast/data_handling/transformers/filters.py index edba07c..94b275f 100644 --- a/src/damast/data_handling/transformers/filters.py +++ b/src/damast/data_handling/transformers/filters.py @@ -42,7 +42,7 @@ def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame: """ mapped_name = self.get_name("x") - df._dataframe = df._dataframe.filter( + df.lazyframe = df.lazyframe.filter( pl.col(mapped_name) != self._remove_value ) return df @@ -61,14 +61,14 @@ def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame: Drop rows with missing value """ mapped_name = self.get_name("x") - dataframe = df._dataframe + dataframe = df.lazyframe new_dataframe = dataframe.drop_nulls(subset=mapped_name) dtype = XDataFrame(new_dataframe).dtype(mapped_name) if dtype not in [str, pl.String]: new_dataframe = new_dataframe.drop_nans(subset=mapped_name) - df._dataframe = new_dataframe + df.lazyframe = new_dataframe return df @@ -96,9 +96,9 @@ def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame: Filter rows and keep those within given values """ mapped_name = self.get_name("x") - dataframe = df._dataframe + dataframe = df.lazyframe new_dataframe = dataframe.filter( pl.col(mapped_name).is_in(self._within_values) ) - df._dataframe = new_dataframe + df.lazyframe = new_dataframe return df diff --git a/src/damast/data_handling/transformers/visualisers.py b/src/damast/data_handling/transformers/visualisers.py index 9943a68..0139a62 100644 --- a/src/damast/data_handling/transformers/visualisers.py +++ b/src/damast/data_handling/transformers/visualisers.py @@ -45,7 +45,7 @@ def __init__(self, @damast.core.output({}) @damast.core.describe("Plot histograms of all columns in dataframe") def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame: - plot_histograms(df=df._dataframe, + plot_histograms(df=df.lazyframe, output_dir=self.output_dir, filename_prefix=self.filename_prefix) return df @@ -63,7 +63,7 @@ def __init__(self, @damast.core.describe("Plot Latitude and longitude ") def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame: - plot_lat_lon(df=df._dataframe, + plot_lat_lon(df=df.lazyframe, output_dir=self.output_dir, filename_prefix=self.filename_prefix) return df diff --git a/src/damast/domains/maritime/transformers/augmenters.py b/src/damast/domains/maritime/transformers/augmenters.py index 5604ecc..09e30aa 100644 --- a/src/damast/domains/maritime/transformers/augmenters.py +++ b/src/damast/domains/maritime/transformers/augmenters.py @@ -84,7 +84,7 @@ def load_data(cls, "y": {"representation_type": float, "unit": damast.core.units.units.deg}}) @damast.core.output({"distance": {"representation_type": float, "unit": damast.core.units.units.km}}) def transform(self, df: damast.core.AnnotatedDataFrame) -> damast.core.AnnotatedDataFrame: - dataframe = df._dataframe + dataframe = df.lazyframe x_name = self.get_name('x') y_name = self.get_name('y') @@ -112,7 +112,7 @@ def transform(self, df: damast.core.AnnotatedDataFrame) -> damast.core.Annotate #dataframe.units[self.get_name("distance")] = damast.core.units.units.km # Drop/hide conversion columns - df._dataframe = dataframe.drop([f"{self.get_name('x')}_rad", f"{self.get_name('y')}_rad"]) + df.lazyframe = dataframe.drop([f"{self.get_name('x')}_rad", f"{self.get_name('y')}_rad"]) return df diff --git a/src/damast/domains/maritime/transformers/features.py b/src/damast/domains/maritime/transformers/features.py index a0b7e14..53f7d0a 100644 --- a/src/damast/domains/maritime/transformers/features.py +++ b/src/damast/domains/maritime/transformers/features.py @@ -51,7 +51,7 @@ def transform(self, """ Compute distance between adjacent messages """ - dataframe = df._dataframe + dataframe = df.lazyframe group = self.get_name("group") in_x = self.get_name("x") @@ -92,7 +92,7 @@ def transform(self, ) # Drop/Hide unused columns - df._dataframe = dataframe.drop([shift_x, shift_y]) + df.lazyframe = dataframe.drop([shift_x, shift_y]) return df @@ -111,12 +111,12 @@ def transform(self, """ Compute distance between adjacent messages """ - dataframe = df._dataframe + dataframe = df.lazyframe delta_distance = self.get_name("delta_distance") delta_time = self.get_name("delta_time") - df._dataframe = dataframe.filter(pl.col(delta_time) != 0).with_columns( + df.lazyframe = dataframe.filter(pl.col(delta_time) != 0).with_columns( (pl.col(delta_distance) / (pl.col(delta_time)/3600.0)).alias("speed") ) return df @@ -141,7 +141,7 @@ def transform(self, """ Compute distance between adjacent messages """ - dataframe = df._dataframe + dataframe = df.lazyframe group = self.get_name("group") sort_column = self.get_name("sort") @@ -184,7 +184,7 @@ def transform(self, (pl.col(delta_heading) / pl.col("_delta_time")).alias("angular_velocity") ).drop("_delta_time") - df._dataframe = dataframe.filter( + df.lazyframe = dataframe.filter( pl.col(delta_heading).is_not_null() ) @@ -215,7 +215,7 @@ def transform(self, pl.col(time).diff().dt.total_seconds().over(group).alias("_delta_time") ) - df._dataframe = dataframe.with_columns( + df.lazyframe = dataframe.with_columns( (pl.col("_delta_heading") / pl.col("_delta_time")).alias("angular_velocity") ).drop("_delta_heading").drop("_delta_time") diff --git a/src/damast/ml/experiments.py b/src/damast/ml/experiments.py index 060f4c8..f6c9e3c 100644 --- a/src/damast/ml/experiments.py +++ b/src/damast/ml/experiments.py @@ -584,7 +584,7 @@ def run(self, filtered_groups = groups_with_sequence_length.filter(pl.col("sequence_length") > self.learning_task.sequence_length) permitted_values = filtered_groups.select(group_column).unique().collect()[:,0] - adf._dataframe = adf.filter(pl.col(group_column).is_in(permitted_values.implode())) + adf.lazyframe = adf.filter(pl.col(group_column).is_in(permitted_values.implode())) features = self.compute_features(adf) train_group, test_group, validate_group = \ diff --git a/tests/damast/cli/test_cli.py b/tests/damast/cli/test_cli.py index dac08fc..8752053 100644 --- a/tests/damast/cli/test_cli.py +++ b/tests/damast/cli/test_cli.py @@ -15,6 +15,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.core.dataframe import DAMAST_SPEC_SUFFIX, AnnotatedDataFrame from damast.domains.maritime.ais.data_generator import AISTestData @@ -27,6 +28,7 @@ def subparsers(): "experiment", "inspect", "process", + "plugins", ] def test_help(subparsers, capsys, monkeypatch): @@ -44,6 +46,7 @@ def test_help(subparsers, capsys, monkeypatch): [ "inspect", DataInspectParser ], [ "experiment", ExperimentParser ], [ "process", DataProcessingParser ], + [ "plugins", PluginsParser ], ]) def test_subparser(name, klass, script_runner): result = script_runner.run(['damast', name, "--help"]) @@ -316,3 +319,32 @@ def test_annotate_representation_type(tmp_path, script_runner): assert result.returncode == 0 +def test_plugins_none_registered(script_runner): + result = script_runner.run(['damast', 'plugins']) + assert result.returncode == 0 + assert re.search("No transformer plugins registered", result.stdout) is not None + + +def test_plugins_lists_registered_entry_point(capsys, monkeypatch): + import importlib.metadata as importlib_metadata + + from damast.core.transformations import PluginManager + + class FakeEntryPoint: + name = "AcmeTransformer" + value = "acme_pkg.transformers:AcmeTransformer" + + def fake_entry_points(*, group): + assert group == PluginManager.ENTRY_POINT_GROUP + return [FakeEntryPoint()] + + monkeypatch.setattr(importlib_metadata, "entry_points", fake_entry_points) + + from damast.cli.plugins import PluginsParser + parser = PluginsParser(parser=ArgumentParser()) + parser.execute(args=None) + + captured = capsys.readouterr() + assert "AcmeTransformer: acme_pkg.transformers:AcmeTransformer" in captured.out + + diff --git a/tests/damast/core/test_dataframe.py b/tests/damast/core/test_dataframe.py index 3fe2fb2..1324148 100644 --- a/tests/damast/core/test_dataframe.py +++ b/tests/damast/core/test_dataframe.py @@ -111,7 +111,7 @@ def test_annotated_dataframe_export_hdf5(metadata, polars_dataframe, tmp_path): assert test_file.exists() with pytest.raises(ValueError, match="no dataframe to save"): - adf._dataframe = None + adf.lazyframe = None adf.save(filename=test_file) loaded_adf = AnnotatedDataFrame.from_file(filename=test_file) @@ -124,7 +124,7 @@ def test_annotated_dataframe_export_hdf5(metadata, polars_dataframe, tmp_path): extra_column = "extra_column" loaded_adf.metadata.columns.append(DataSpecification(name=extra_column)) from_column = loaded_adf.column_names[0] - loaded_adf._dataframe = loaded_adf._dataframe.with_columns( + loaded_adf.lazyframe = loaded_adf.lazyframe.with_columns( polars.col(from_column).alias(extra_column) ) loaded_adf.save(filename=test_file) @@ -190,7 +190,7 @@ def test_annotated_dataframe_import_csv(data_path): assert adf.column_names == ["height", "letter"] assert adf.dtype('height') == polars.Int64, "None types should be properly handled" - assert XDataFrame(adf._dataframe).equals(XDataFrame(polars.scan_csv(csv_path, null_values=["None", "none"]))) + assert XDataFrame(adf.lazyframe).equals(XDataFrame(polars.scan_csv(csv_path, null_values=["None", "none"]))) assert adf._metadata.annotations["license"] == Annotation(name="license", value="MIT License") assert adf._metadata.annotations["comment"] == Annotation(name="comment", value="test dataframe") assert adf._metadata.columns[0] == DataSpecification( @@ -208,7 +208,7 @@ def test_annotated_dataframe_import_csv_with_quotes(data_path): assert adf.dtype('id') == polars.Int64 assert adf.dtype('name') == polars.String - assert XDataFrame(adf._dataframe).equals(XDataFrame(polars.scan_csv(csv_path, null_values=["None", "none"]))) + assert XDataFrame(adf.lazyframe).equals(XDataFrame(polars.scan_csv(csv_path, null_values=["None", "none"]))) df = adf.dataframe.collect() assert df[0,1] == "a,b;c" @@ -249,7 +249,7 @@ def test_01_dataframe_composition(data_path): metadata=md) assert adf._metadata == md - assert XDataFrame(adf._dataframe).equals(XDataFrame(df)) + assert XDataFrame(adf.lazyframe).equals(XDataFrame(df)) assert adf.column_names == df.compat.column_names @@ -281,7 +281,7 @@ def test_force_range(): validation_mode=ValidationMode.UPDATE_DATA ) - assert XDataFrame(adf._dataframe).equals(XDataFrame(df_filtered)) + assert XDataFrame(adf.lazyframe).equals(XDataFrame(df_filtered)) def test_convert_csv_to_adf(tmp_path): output_filename = tmp_path / "test-convert-csv.pq" diff --git a/tests/damast/core/test_dataprocessing.py b/tests/damast/core/test_dataprocessing.py index 1587a0d..8476fcd 100644 --- a/tests/damast/core/test_dataprocessing.py +++ b/tests/damast/core/test_dataprocessing.py @@ -108,10 +108,10 @@ class TransformerB(PipelineElement): def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame: # This operation is does not really make sense, but acts as a placeholder to generate # the desired output columns - df._dataframe = df.with_columns( + df.lazyframe = df.with_columns( (polars.col("longitude_x") - polars.col("longitude_y")).alias("delta_longitude") ) - df._dataframe = df.with_columns( + df.lazyframe = df.with_columns( (polars.col("latitude_x") - polars.col("latitude_y")).alias("delta_latitude") ) return df @@ -129,7 +129,7 @@ class TransformerC(PipelineElement): "label": {} }) def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame: - df._dataframe = df._dataframe.with_columns( + df.lazyframe = df.lazyframe.with_columns( polars.lit("data-label").alias("label") ) return df @@ -158,7 +158,7 @@ def transform(self, df: AnnotatedDataFrame, other: AnnotatedDataFrame) -> Annota 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) df._metadata = df._metadata.merge(other._metadata).drop(other_timestamp) return df @@ -205,7 +205,7 @@ def transform(self, df: AnnotatedDataFrame, other: AnnotatedDataFrame) -> Annota other_timestamp = self.get_name('timestamp', datasource='other') df_timestamp = self.get_name('timestamp') - filtered_df = df.join_where(other._dataframe, \ + filtered_df = df.join_where(other.lazyframe, \ (pl.col(df_timestamp) - self.before_time_in_s) <= pl.col(other_timestamp), \ (pl.col(df_timestamp) + self.after_time_in_s) >= pl.col(other_timestamp), \ great_circle_distance(pl.col(self.get_name('lat')), @@ -213,14 +213,14 @@ def transform(self, df: AnnotatedDataFrame, other: AnnotatedDataFrame) -> Annota pl.col(self.get_name('lat', datasource='other')), pl.col(self.get_name('lon', datasource='other'))) <= self.distance_in_km ) - df._dataframe = df.join(filtered_df, + df.lazyframe = df.join(filtered_df, how="left", left_on=[self.get_name('mmsi'), df_timestamp], right_on=[self.get_name('mmsi'), df_timestamp], suffix="_redundant", ).drop(cs.ends_with("_redundant")) - df._dataframe = df.with_columns( + df.lazyframe = df.with_columns( event_delta_distance = great_circle_distance(pl.col(self.get_name('lat')), pl.col(self.get_name('lon')), pl.col(self.get_name('lat', datasource='other')), @@ -448,7 +448,7 @@ class TransformX(PipelineElement): @damast.core.input({"x": {"unit": units.deg}}) @damast.core.output({"{{x}}_suffix": {"unit": units.deg}}) def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame: - df._dataframe = df._dataframe.with_columns(polars.col(self.get_name('x')).alias(f"{self.get_name('x')}_suffix")) + df.lazyframe = df.lazyframe.with_columns(polars.col(self.get_name('x')).alias(f"{self.get_name('x')}_suffix")) return df pipeline = DataProcessingPipeline(name="TransformStatus", diff --git a/tests/damast/core/test_transformations.py b/tests/damast/core/test_transformations.py new file mode 100644 index 0000000..0a5b838 --- /dev/null +++ b/tests/damast/core/test_transformations.py @@ -0,0 +1,193 @@ +import importlib.metadata +import os +import sys + +import pytest + +import damast +from damast.core.transformations import ( + PipelineElement, + PluginManager, + plugin_manager, +) +from damast.data_handling.transformers.cycle_transformer import CycleTransformer + +LOCAL_TRANSFORMER_SOURCE = """ +from damast.core.transformations import PipelineElement +from damast.core.dataframe import AnnotatedDataFrame +from damast.core.decorators import describe, input, output + + +class LocalDoubler(PipelineElement): + @describe("doubles a column") + @input({"x": {}}) + @output({"x_doubled": {}}) + def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame: + return df +""" + + +def _reset_plugin_manager(): + for module_name in list(plugin_manager.local_files): + sys.modules.pop(module_name, None) + plugin_manager._local_modules.clear() + plugin_manager._local_files.clear() + plugin_manager._requirement_cache.clear() + plugin_manager._loaded = False + + +@pytest.fixture +def local_plugin_path(tmp_path, monkeypatch): + plugin_dir = tmp_path / "plugins" + plugin_dir.mkdir() + monkeypatch.setenv(PluginManager.PLUGIN_PATH_ENV, str(plugin_dir)) + + yield plugin_dir + + _reset_plugin_manager() + +# CycleTransformer is a built-in PipelineElement whose module lives inside the +# installed 'damast' distribution - used here as a stand-in for a transformer +# provided by any installed (plugin) package. + + +def test_resolve_requirement_for_installed_package(): + requirement = plugin_manager.resolve_requirement(CycleTransformer.__module__) + assert requirement == {"distribution": "damast", "version": damast.version.__version__} + + +def test_resolve_requirement_unresolvable_module(): + assert plugin_manager.resolve_requirement("this_module_does_not_exist_anywhere") is None + + +def test_pipeline_element_iter_includes_requires(): + data = dict(CycleTransformer(n=1)) + assert data["requires"] == {"distribution": "damast", "version": damast.version.__version__} + + +def test_create_new_roundtrip_with_matching_requirement(): + step = dict(CycleTransformer(n=1)) + instance = PipelineElement.create_new(**step) + assert isinstance(instance, CycleTransformer) + + +def test_create_new_missing_plugin_package_raises_actionable_error(): + step = dict(CycleTransformer(n=1)) + step["requires"] = {"distribution": "acme-damast-plugins", "version": "1.2.3"} + + with pytest.raises(ImportError, match="pip install acme-damast-plugins==1.2.3"): + PipelineElement.create_new(**step) + + +def test_create_new_version_mismatch_warns_but_loads(caplog): + step = dict(CycleTransformer(n=1)) + step["requires"] = {"distribution": "damast", "version": "0.0.0-does-not-match"} + + with caplog.at_level("WARNING"): + instance = PipelineElement.create_new(**step) + + assert isinstance(instance, CycleTransformer) + assert any("0.0.0-does-not-match" in record.message for record in caplog.records) + + +def test_list_plugins_discovers_entry_points(monkeypatch): + class FakeEntryPoint: + name = "AcmeTransformer" + value = "acme_pkg.transformers:AcmeTransformer" + + def fake_entry_points(*, group): + assert group == PluginManager.ENTRY_POINT_GROUP + return [FakeEntryPoint()] + + monkeypatch.setattr(importlib.metadata, "entry_points", fake_entry_points) + + assert PipelineElement.list_plugins() == {"AcmeTransformer": "acme_pkg.transformers:AcmeTransformer"} + + +def test_list_plugins_empty_by_default(): + assert PipelineElement.list_plugins() == {} + + +def test_local_plugin_path_discovered_via_list_plugins(local_plugin_path): + (local_plugin_path / "acme_local_transformer.py").write_text(LOCAL_TRANSFORMER_SOURCE) + PipelineElement.reload_plugins() + + plugins = PipelineElement.list_plugins() + assert plugins["LocalDoubler"] == "acme_local_transformer:LocalDoubler" + + +def test_local_plugin_path_resolvable_via_create_new(local_plugin_path): + (local_plugin_path / "acme_local_transformer2.py").write_text(LOCAL_TRANSFORMER_SOURCE) + PipelineElement.reload_plugins() + + instance = PipelineElement.create_new(module_name="acme_local_transformer2", class_name="LocalDoubler") + assert instance.__class__.__name__ == "LocalDoubler" + + # a locally-loaded transformer is not backed by an installable distribution, but is + # flagged via a 'hint' so it is clear (and traceable) that it came from DAMAST_PLUGIN_PATH + expected_path = local_plugin_path / "acme_local_transformer2.py" + assert dict(instance)["requires"] == {"hint": "local", "path": str(expected_path)} + + +def test_missing_local_plugin_error_mentions_original_path(local_plugin_path): + plugin_file = local_plugin_path / "acme_local_transformer3.py" + plugin_file.write_text(LOCAL_TRANSFORMER_SOURCE) + PipelineElement.reload_plugins() + + step = PipelineElement.create_new(module_name="acme_local_transformer3", class_name="LocalDoubler") + saved_step = dict(step) + assert saved_step["requires"] == {"hint": "local", "path": str(plugin_file)} + + # simulate loading the pipeline elsewhere, where this local plugin file is unavailable + sys.modules.pop("acme_local_transformer3", None) + plugin_manager._local_modules.pop("acme_local_transformer3", None) + plugin_manager._local_files.pop("acme_local_transformer3", None) + + with pytest.raises(ImportError, match="originally loaded from"): + PipelineElement.create_new(**saved_step) + + +def test_local_plugin_path_missing_directory_warns_but_does_not_crash(tmp_path, monkeypatch, caplog): + missing_dir = tmp_path / "does-not-exist" + monkeypatch.setenv(PluginManager.PLUGIN_PATH_ENV, str(missing_dir)) + + with caplog.at_level("WARNING"): + PipelineElement.reload_plugins() + + assert PipelineElement.list_plugins() == {} + assert any("is not a directory" in record.message for record in caplog.records) + + _reset_plugin_manager() + + +def test_local_plugin_path_name_collision_warns_and_keeps_first(tmp_path, monkeypatch, caplog): + dir_a = tmp_path / "a" + dir_b = tmp_path / "b" + dir_a.mkdir() + dir_b.mkdir() + (dir_a / "same_name.py").write_text(LOCAL_TRANSFORMER_SOURCE) + (dir_b / "same_name.py").write_text(LOCAL_TRANSFORMER_SOURCE) + + monkeypatch.setenv(PluginManager.PLUGIN_PATH_ENV, os.pathsep.join([str(dir_a), str(dir_b)])) + + with caplog.at_level("WARNING"): + PipelineElement.reload_plugins() + + assert any("collides with" in record.message for record in caplog.records) + assert plugin_manager.local_files["same_name"] == dir_a / "same_name.py" + + _reset_plugin_manager() + + +def test_create_new_missing_local_module_error_mentions_plugin_path(monkeypatch): + monkeypatch.setenv(PluginManager.PLUGIN_PATH_ENV, "/some/configured/path") + + with pytest.raises(ImportError, match=PluginManager.PLUGIN_PATH_ENV): + PipelineElement.create_new(module_name="totally_missing_local_module", class_name="Foo") + + +def test_plugin_manager_is_a_separate_instantiable_class(): + manager = PluginManager() + assert manager is not plugin_manager + assert manager.list_plugins() == {} + assert manager.local_files == {} diff --git a/tests/damast/domains/maritime/ais/test_augmenters.py b/tests/damast/domains/maritime/ais/test_augmenters.py index 8ad550d..1faa481 100644 --- a/tests/damast/domains/maritime/ais/test_augmenters.py +++ b/tests/damast/domains/maritime/ais/test_augmenters.py @@ -126,7 +126,7 @@ def test_delta_column(tmp_path): for mmsi, data in df_grouped: # group specific distances in the result - distances = new_adf._dataframe.filter(pl.col(ColumnName.MMSI) == mmsi[0]).select(ColumnName.DELTA_DISTANCE).collect().to_numpy() + distances = new_adf.lazyframe.filter(pl.col(ColumnName.MMSI) == mmsi[0]).select(ColumnName.DELTA_DISTANCE).collect().to_numpy() # original data in df, so check against expected_dataframe = data.select( diff --git a/tests/examples/01-ais-ml/ais_processing.py b/tests/examples/01-ais-ml/ais_processing.py index 2c8e8ad..20975de 100644 --- a/tests/examples/01-ais-ml/ais_processing.py +++ b/tests/examples/01-ais-ml/ais_processing.py @@ -57,7 +57,7 @@ def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame: _df = lat_cyclic_transformer.fit_transform(df=df) _df = lon_cyclic_transformer.fit_transform(df=_df) - df._dataframe = _df + df.lazyframe = _df return df @@ -71,7 +71,7 @@ def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame: new_df = pipeline.transform(adf) print(pipeline.to_str(indent_level=2)) -print(new_df._dataframe.collect()) +print(new_df.lazyframe.collect()) # Start ML # ml = MLPipeline(name="train") diff --git a/tox.ini b/tox.ini index 32c79fd..6ccbd66 100644 --- a/tox.ini +++ b/tox.ini @@ -64,6 +64,7 @@ commands = bash -c "damast convert --help > {toxinidir}/docs/examples/damast-convert-help.txt" bash -c "damast annotate --help > {toxinidir}/docs/examples/damast-annotate-help.txt" bash -c "damast process --help > {toxinidir}/docs/examples/damast-process-help.txt" + bash -c "damast plugins --help > {toxinidir}/docs/examples/damast-plugins-help.txt" jupyter book build . [flake8]