[WIP] metadata: various fixes

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_min_max(adf._dataframe, column_spec.name)
+                    metadata[column_spec.name].update_datarange_and_stats(adf._dataframe, column_spec.name)
 
         adf._metadata = metadata
         adf.validate_metadata(ValidationMode.UPDATE_DATA)

src/damast/cli/data_inspect.py

@@ -8,6 +8,7 @@
 import damast  # noqa
 from damast.cli.base import BaseParser
 from damast.core.dataframe import AnnotatedDataFrame
+from damast.core.metadata import ValidationMode
 from damast.utils.io import Archive
 
 logger = logging.getLogger(__name__)
@@ -45,6 +46,11 @@ def __init__(self, parser: ArgumentParser):
                 required=False
         )
 
+        parser.add_argument("--validation-mode",
+                            default="readonly",
+                            choices=[x.value.lower() for x in ValidationMode],
+                            help="Define the validation mode")
+
     def expand_filter_arg(self, adf: AnnotatedDataFrame, arg: str):
         if arg in adf.column_names:
             return f"pl.col('{arg}')"
@@ -69,7 +75,13 @@ def execute(self, args):
                 if not files:
                     raise RuntimeError(f"Inspection is not supported for input files: {input_files=}")
 
-                adf = AnnotatedDataFrame.from_files(files=files, metadata_required=False)
+                try:
+                    validation_mode = ValidationMode[args.validation_mode.upper()]
+                except KeyError:
+                    raise ValueError(f"--validation-mode has invalid argument."
+                                     f" Select from: {[x.value.lower() for x in ValidationMode]}")
+
+                adf = AnnotatedDataFrame.from_files(files=files, metadata_required=False, validation_mode=validation_mode)
 
                 if args.filter:
                     filter_values = ""

src/damast/cli/main.py

@@ -1,6 +1,7 @@
 """
 Main argument parser and CLI entry point.
 """
+import logging
 import sys
 import traceback as tb
 from argparse import ArgumentParser
@@ -13,7 +14,20 @@
 from damast.cli.data_inspect import DataInspectParser
 from damast.cli.data_processing import DataProcessingParser
 from damast.cli.experiment import ExperimentParser
+from damast.config import DAMAST_LOG_DATE_FORMAT, DAMAST_LOG_FORMAT, DAMAST_LOG_STYLE
 
+logging.basicConfig(
+    format=DAMAST_LOG_FORMAT,
+    style=DAMAST_LOG_STYLE,
+    datefmt=DAMAST_LOG_DATE_FORMAT
+)
+
+# Safely get the mapping dictionary regardless of Python version
+if hasattr(logging, "getLevelNamesMapping"):
+    level_mapping = logging.getLevelNamesMapping()
+else:
+    # Fallback for Python 3.10 and older
+    level_mapping = getattr(logging, "_nameToLevel", {})
 
 class MainParser(ArgumentParser):
 
@@ -23,8 +37,11 @@ def __init__(self, **kwargs):
 
         self.add_argument("-w", "--workdir", default=str(Path(".").resolve()))
         self.add_argument("-v", "--verbose", action="store_true")
-        self.add_argument("--loglevel", dest="loglevel", type=int, default=10, help="Set loglevel to display")
-        self.add_argument("--logfile", dest="logfile", type=str, default=None,
+        self.add_argument("--log-level", type=str,
+                          default="INFO",
+                          choices=[x for x in level_mapping],
+                          help="Set loglevel to display")
+        self.add_argument("--log-file", type=str, default=None,
                           help="Set file for saving log (default prints to terminal)")
 
         self.add_argument("--version", action="store_true", default=False,
@@ -76,6 +93,10 @@ def run():
         print(f"damast {damast_version}")
         sys.exit(0)
 
+    for current_logger in [logging.getLogger(x) for x in logging.root.manager.loggerDict]:
+        if current_logger.name.startswith("damast"):
+            current_logger.setLevel(logging.getLevelName(args.log_level))
+
     if hasattr(args, "active_subparser"):
         try:
             args.active_subparser.execute(args)

src/damast/config.py

@@ -0,0 +1,3 @@
+DAMAST_LOG_FORMAT = '[{asctime}][{levelname:^8s}] {name}: {message}'
+DAMAST_LOG_STYLE = '{'
+DAMAST_LOG_DATE_FORMAT = '%Y-%m-%d %H:%M:%S'

src/damast/core/data_description.py

@@ -3,6 +3,8 @@
 """
 from __future__ import annotations
 
+import datetime as dt
+import logging
 import math
 from abc import ABC, abstractmethod
 from typing import Any, Dict, List
@@ -13,6 +15,7 @@
 
 __all__ = ["CyclicMinMax", "DataElement", "DataRange", "ListOfValues", "MinMax"]
 
+logger = logging.getLogger(__name__)
 
 class DataElement:
     """
@@ -27,14 +30,17 @@ def create(cls, value, dtype):
         :param value: value of the DataElement
         :param dtype: datatype of the value
         """
-        if isinstance(dtype, pl.datatypes.classes.DataTypeClass):
-            dtype = dtype.to_python()
-        elif isinstance(dtype, pl.datatypes.Datetime):
-            if type(value) is str:
-                return pl.Series([value]).str.to_datetime(time_unit=dtype.time_unit, time_zone=dtype.time_zone)[0]
-            return pl.Series([value]).cast(dtype)[0]
-
-        return dtype(value)
+        try:
+            if isinstance(dtype, pl.datatypes.classes.DataTypeClass):
+                dtype = dtype.to_python()
+            elif isinstance(dtype, pl.datatypes.Datetime) or (dtype is dt.datetime):
+                if type(value) is str:
+                    return pl.Series([value]).str.to_datetime(time_unit=dtype.time_unit, time_zone=dtype.time_zone)[0]
+                return pl.Series([value]).cast(dtype)[0]
+            return dtype(value)
+        except Exception as e:
+            logger.warning(f"DataElement.create: {value=} {dtype=} failed -- {e}")
+            raise
 
 
 class DataRange(ABC):
@@ -126,7 +132,7 @@ class ListOfValues:
 
     def __init__(self, values: List[Any]):
         """
-        Initialise ListOfValue
+        Initialise ListOfValues
 
         :param values: values that define this list
         :raise ValueError: Raises if values is not a list.
@@ -197,7 +203,13 @@ def to_dict(self) -> Dict[str, Any]:
         """
         return dict(self)
 
-    def merge(self, other: ListOfValues) -> ListOfValues:
+    def merge(self, other: ListOfValues | MinMax) -> ListOfValues:
+        if type(other) is MinMax:
+            this_min_max = MinMax(min(self.values), max(self.values))
+            return this_min_max.merge(other)
+        elif type(other) is not ListOfValues:
+            raise ValueError(f"ListOfValues.merge: cannot merge with other (type: {type(other)})")
+
         self.values = list(set(self.values + other.values))
         self.values.sort(key=lambda e: (e is None, e))
         return self
@@ -257,6 +269,10 @@ def is_in_range(self, value: Any) -> bool:
         except Exception:
             pass
 
+        if type(value) is float:
+            epsilon = 1.0E-07
+            return self.min - epsilon <= value <= self.max + epsilon
+
         return self.min <= value <= self.max
 
     @classmethod

src/damast/core/dataframe.py

@@ -209,12 +209,16 @@ def get_supported_format(cls, suffix: str) -> str | None:
     def from_files(cls,
             files: list[str|Path],
             metadata_required: bool = True,
-            validation_mode: ValidationMode = ValidationMode.READONLY
+            validation_mode: ValidationMode = ValidationMode.READONLY,
+            merge_strategy: DataSpecification.MergeStrategy = DataSpecification.MergeStrategy.THIS
         ) -> AnnotatedDataFrame:
         """
         Create an annotated dataframe by loading given files
 
         :param files: Files to use for importing and creating the annotated dataframe
+        :param metadata_required metadata needs to be available either as spec.yaml file or embedded into the file
+        :param validation_mode metadata will be validated, updated or ignored according to this mode
+        :param merge_strategy for combining the metadata of multiple files, select the merge strategy
 
         """
         metadata = None
@@ -282,7 +286,7 @@ def from_files(cls,
                         annotations=list(metadata_list[0].annotations.values())
                     )
             for i in range(0, len(metadata_list)-1):
-                metadata = metadata.merge(metadata_list[i+1])
+                metadata = metadata.merge(metadata_list[i+1], strategy=merge_strategy)
         else:
             for _, m in metadata.items():
                 metadata = m
@@ -293,18 +297,26 @@ def from_files(cls,
     @classmethod
     def load_parquet(cls, files) -> tuple[AnnotatedDataFrame, dict[str, MetaData]]:
             _log.info(f"Loading parquet: {files=}")
-            df = polars.scan_parquet(files, missing_columns='insert')
-
             metadata_per_file = {}
+
+            pyarrow_schemas = []
             for file in files:
                 schema = pq.read_schema(file)
+                pyarrow_schemas.append(schema)
                 if schema and hasattr(schema, "metadata"):
                     if schema.metadata is not None:
                         if b"annotated_dataframe" in schema.metadata:
                             data = schema.metadata[b"annotated_dataframe"]
                             m = MetaData.from_dict(json.loads(data.decode('UTF-8')))
                             m.set_annotation(Annotation(name=Annotation.Key.Source, value=Path(file).name))
                             metadata_per_file[file] = m
+
+            # https://github.com/pola-rs/polars/issues/27280
+            arrow_schema = pyarrow.unify_schemas(pyarrow_schemas)
+            sorted_arrow_schema = pyarrow.schema(sorted(arrow_schema, key=lambda x: x.name))
+            polars_schema = polars.from_arrow(sorted_arrow_schema.empty_table()).schema
+
+            df = polars.scan_parquet(files, missing_columns='insert', schema=polars_schema)
             return df, metadata_per_file
 
     @classmethod
@@ -332,14 +344,19 @@ def load_csv(cls, files) -> tuple[AnnotatedDataFrame, dict[str, MetaData]]:
     def from_file(cls,
             filename: str | Path,
             metadata_required: bool = True,
+            validation_mode: ValidationMode = ValidationMode.READONLY,
+            merge_strategy: DataSpecification.MergeStrategy = DataSpecification.MergeStrategy.THIS
         ) -> AnnotatedDataFrame:
         """
         Create an annotated dataframe from an hdf5 file.
 
         :param filename: Filename to use for importing and creating the annotated dataframe
 
         """
-        return cls.from_files([filename], metadata_required=metadata_required)
+        return cls.from_files([filename],
+                              metadata_required=metadata_required,
+                              validation_mode=validation_mode,
+                              merge_strategy=merge_strategy)
 
     @classmethod
     def infer_annotation(cls, df: DataFrame) -> MetaData:
@@ -426,7 +443,7 @@ def convert_csv_to_adf(
 
     def drop(self, columns, strict: bool = True) -> AnnotatedDataFrame:
         self._dataframe = self._dataframe.drop(columns, strict=strict)
-        self._metadata.drop(columns)
+        self._metadata = self._metadata.drop(columns)
         return self
 
     def copy(self):
@@ -437,4 +454,5 @@ def shape(self):
         return self._dataframe.compat.collected().shape
 
     def __deepcopy__(self, memo=None):
-        return AnnotatedDataFrame(self._dataframe.clone(), copy.deepcopy(self._metadata))
+        # ignore validation, also erroneous frames should be copyable
+        return AnnotatedDataFrame(self._dataframe.clone(), copy.deepcopy(self._metadata), validation_mode=ValidationMode.IGNORE)

src/damast/core/dataprocessing.py

@@ -477,7 +477,6 @@ def transform(self,
 
         adf = pipeline._run(in_dataframes, verbose=verbose)
         assert isinstance(adf, AnnotatedDataFrame)
-
         adf.validate_metadata(validation_mode=damast.core.ValidationMode.UPDATE_METADATA)
         return adf
 
@@ -521,7 +520,7 @@ def _run(self,
 
     def on_transform_start(self,
                            step: PipelineElement,
-                           adf: AnnotatedDataFrame):
+                           dataframes: dict[str, AnnotatedDataFrame]):
         """
         Default implementation of the on_transform_start callback.
 
@@ -536,7 +535,7 @@ def on_transform_start(self,
 
         step_name = self.processing_graph[step.uuid].name
         self._processing_stats[step_name] = {
-            "input_dataframe_length": adf.shape[0],
+            "input_dataframe_length": {x: y.shape[0] for x,y in dataframes.items()},
             "start_time": start_time
         }
 

src/damast/core/decorators.py

@@ -45,6 +45,25 @@ def _get_dataframe(*args, **kwargs) -> AnnotatedDataFrame:
 
     return datasource
 
+def _get_dataframes(*args, **kwargs) -> AnnotatedDataFrame:
+    """
+    Extract all dataframes from positional or keyword arguments
+    :param datasource: the expected name of the datasource
+    :param args: positional arguments
+    :param kwargs: keyword arguments
+    :return: The annotated data frame
+    :raise KeyError: if a positional argument does not exist and keyword :code:`df` is missing
+    :raise TypeError: if the kwargs 'df' is not an AnnotatedDataFrame
+    """
+    arguments = kwargs.copy()
+    arg_index = 0
+    for parameter in inspect.signature(args[0].transform).parameters:
+        if parameter not in kwargs:
+            arg_index += 1
+            arguments[parameter] = args[arg_index]
+
+    return {x: y for x,y in arguments.items() if isinstance(y, AnnotatedDataFrame)}
+
 def describe(description: str):
     """
     Specify the description for the transformation for the decorated function.
@@ -109,7 +128,8 @@ def check(*args, **kwargs):
                     # if this is the last datasource parameter, this is also the last input decorators
                     # so the transform can start
                     if label == parameters[-1]:
-                        getattr(pipeline_element, "parent_pipeline").on_transform_start(pipeline_element, adf=_df)
+                        dataframes = _get_dataframes(*args, **kwargs)
+                        getattr(pipeline_element, "parent_pipeline").on_transform_start(pipeline_element, dataframes=dataframes)
                 return func(*args, **kwargs)
 
             raise RuntimeError(
@@ -180,6 +200,7 @@ def check(*args, **kwargs) -> AnnotatedDataFrame:
             try:
                 # Ensure that metadata is up to date with the dataframe
                 adf.update(expectations=pipeline_element.output_specs)
+                adf.validate_metadata()
             except RuntimeError as e:
                 txt = f"Failed to update metadata in pipeline element: {pipeline_element}"
                 if parent_pipeline: