From f35fe6f8e476167db875630191754f5dfb90bd01 Mon Sep 17 00:00:00 2001 From: Thomas Roehr Date: Thu, 2 Jul 2026 15:29:08 +0200 Subject: [PATCH 1/6] metadata: support setting metadata and loading files with overlapping metadata --- src/damast/core/data_description.py | 10 ++- src/damast/core/dataframe.py | 20 ++++-- src/damast/core/metadata.py | 102 +++++++++++++++++++--------- src/damast/core/polars_dataframe.py | 1 + 4 files changed, 93 insertions(+), 40 deletions(-) diff --git a/src/damast/core/data_description.py b/src/damast/core/data_description.py index 51d9bb8..772c243 100644 --- a/src/damast/core/data_description.py +++ b/src/damast/core/data_description.py @@ -126,7 +126,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 +197,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 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 diff --git a/src/damast/core/dataframe.py b/src/damast/core/dataframe.py index 0aa3aa8..d495601 100644 --- a/src/damast/core/dataframe.py +++ b/src/damast/core/dataframe.py @@ -209,7 +209,8 @@ 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 @@ -282,7 +283,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,11 +294,12 @@ 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: @@ -305,6 +307,13 @@ def load_parquet(cls, files) -> tuple[AnnotatedDataFrame, dict[str, MetaData]]: 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,6 +341,7 @@ def load_csv(cls, files) -> tuple[AnnotatedDataFrame, dict[str, MetaData]]: def from_file(cls, filename: str | Path, metadata_required: bool = True, + merge_strategy: DataSpecification.MergeStrategy = DataSpecification.MergeStrategy.THIS ) -> AnnotatedDataFrame: """ Create an annotated dataframe from an hdf5 file. @@ -339,7 +349,7 @@ def from_file(cls, :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, merge_strategy=merge_strategy) @classmethod def infer_annotation(cls, df: DataFrame) -> MetaData: diff --git a/src/damast/core/metadata.py b/src/damast/core/metadata.py index 6a66932..0c3c7a6 100644 --- a/src/damast/core/metadata.py +++ b/src/damast/core/metadata.py @@ -20,7 +20,13 @@ from .annotations import Annotation, History from .constants import DAMAST_HDF5_COLUMNS, DAMAST_HDF5_ROOT, DAMAST_SPEC_SUFFIX -from .data_description import DataElement, DataRange, MinMax, NumericValueStats +from .data_description import ( + DataElement, + DataRange, + ListOfValues, + MinMax, + NumericValueStats, +) from .formatting import DEFAULT_INDENT from .types import DataFrame, XDataFrame from .units import Unit, units @@ -603,7 +609,7 @@ def apply( xdf = XDataFrame(df) if self.representation_type is not None: dtype = xdf.dtype(column_name) - if dtype != self.representation_type: + if self.representation_type not in [dtype, dtype.to_python()]: warnings.warn( f"{self.__class__.__name__}.apply: column '{column_name}':" f" expected representation type: {self.representation_type}," @@ -633,23 +639,35 @@ def apply( return xdf._dataframe if validation_mode == ValidationMode.UPDATE_METADATA: - return self.update_min_max(df, column_name) + return self.update_datarange_and_stats(df, column_name) - def update_min_max(self, df, column_name): + def update_datarange_and_stats(self, df, column_name: str): xdf = XDataFrame(df) - self.representation_type = xdf.dtype(column_name) + if df.compat.is_string(column_name): + categories = df.compat.categories(column_name) + if categories: + logger.debug(f"Setting value range: ListOfValues for {column_name}") + self.value_range = ListOfValues(categories) + else: + logger.debug(f"Setting value range: MinMax for {column_name}") + min_value, max_value = df.compat.minmax(column_name) + if min_value is not None and max_value is not None: + self.value_range = MinMax(min_value, max_value) + elif df.compat.is_numeric(column_name): + try: + min_value, max_value = df.compat.minmax(column_name) + if min_value is not None and max_value is not None: + logger.debug(f"Setting value range: MinMax for {column_name}") + self.value_range = MinMax(min_value, max_value) - try: - min_value, max_value = xdf.minmax(column_name) - logger.info( - f"Setting MinMax range ({min_value}, {max_value}) for {column_name}" - ) - self.value_range = MinMax(min_value, max_value) - except ValueError: - # Type might not be numeric - pass - return xdf._dataframe + results = df.compat.minmax_stats([column_name]) + logger.debug(f"Setting value stats for {column_name}") + self.value_stats = results[column_name]["stats"] + except ValueError as e: + logger.debug(f"Metadata.update_datarange_and_stats: could not update datarange and stats '{column_name}' -- {e}") + + return xdf._dataframe def get_fulfillment(self, data_spec: DataSpecification) -> Fulfillment: """ @@ -766,6 +784,7 @@ def merge(self, other: DataSpecification, strategy: MergeStrategy | None = None) ) ds = DataSpecification(name=self.name) + for key in self.Key: if key == self.Key.name: continue @@ -786,22 +805,27 @@ def merge(self, other: DataSpecification, strategy: MergeStrategy | None = None) elif this_value == other_value: setattr(ds, key.value, this_value) else: - if hasattr(this_value, "merge"): - merged_value = this_value.merge(other_value) - setattr(ds, key.value, merged_value) - elif strategy: - if strategy == DataSpecification.MergeStrategy.OTHER: - setattr(ds, key.value, other_value) - elif strategy == DataSpecification.MergeStrategy.THIS: - setattr(ds, key.value, this_value) - else: - raise RuntimeError(f"DataSpecification.merge: Invalid merge strategy {strategy} provided") - else: - raise ValueError( - f"{self.__class__.__name__}.merge cannot merge specs for '{self.name}': value for '{key.value}' differs: " + try: + if hasattr(this_value, "merge"): + merged_value = this_value.merge(other_value) + setattr(ds, key.value, merged_value) + return ds + except Exception as e: + if not strategy: + raise ValueError( + f"{self.__class__.__name__}.merge cannot merge specs for '{self.name}': value for '{key.value}' differs: " - f" on self: '{this_value}' vs. other: '{other_value}'" - ) + f" on self: '{this_value}' vs. other: '{other_value}' -- {e}" + ) + + logger.info(f"{self.__class__.__name__}.merge: using merge strategy {strategy} for {key.value}: this={this_value} -- other={other_value}") + + if strategy == DataSpecification.MergeStrategy.OTHER: + setattr(ds, key.value, other_value) + elif strategy == DataSpecification.MergeStrategy.THIS: + setattr(ds, key.value, this_value) + else: + raise RuntimeError(f"{self.__class__.__name__}.merge: Invalid merge strategy {strategy} provided") return ds @classmethod @@ -828,8 +852,11 @@ def merge_lists( if column_name in b_column_dict: # Need to check merge b_column_spec = b_column_dict[column_name] - - merged_spec = a_spec.merge(other=b_column_spec, strategy=strategy) + try: + merged_spec = a_spec.merge(other=b_column_spec, strategy=strategy) + except Exception as e: + raise RuntimeError(f"{cls.__name__}.merge_lists: cannot merge spec for column: '{column_name}'.\n" + f"a={a_spec} to be merged with b={b_column_spec} -- {e}") result_specs.append(merged_spec) else: result_specs.append(a_spec) @@ -1149,8 +1176,8 @@ def apply( """ assert isinstance(df, pl.LazyFrame), f"Expected polars.LazyFrame, got {type(df)}" + columns = df.compat.column_names for column_spec in self.columns: - columns = df.compat.column_names if column_spec.name in columns: try: df = column_spec.apply( @@ -1167,6 +1194,15 @@ def apply( f"{self.__class__.__name__}.apply: missing column '{column_spec.name}' in dataframe." f" Found {len(columns)} column(s): {','.join(columns)}" ) + + if validation_mode != ValidationMode.IGNORE: + delta = set(columns).difference([x.name for x in self.columns]) + if delta: + raise ValueError( + f"{self.__class__.__name__}.apply: missing column metatadata for columns: {','.join(delta)}. " + f"Column data exists in dataframe, but no metadata is available" + ) + return df def __contains__(self, column_name: str): diff --git a/src/damast/core/polars_dataframe.py b/src/damast/core/polars_dataframe.py index aa8427d..5dce0d7 100644 --- a/src/damast/core/polars_dataframe.py +++ b/src/damast/core/polars_dataframe.py @@ -167,6 +167,7 @@ def categories(self, column_name: str, max_count: int = 100) -> list[str]: raise RuntimeError(f"Failed to extract categories for column '{column_name}' -- {e}") from e if len(categories) <= max_count: + # do not count every timepoint as category timepoint_like = 0 for c in categories[:10]: if c is None: From 91447d06eab4ced32f129d3722f1d0ed8920d226 Mon Sep 17 00:00:00 2001 From: Thomas Roehr Date: Mon, 6 Jul 2026 16:22:52 +0200 Subject: [PATCH 2/6] Update metadata handling to account for stricter mode --- src/damast/cli/data_annotate.py | 2 +- src/damast/cli/data_inspect.py | 14 +++++++- src/damast/core/dataframe.py | 12 +++++-- src/damast/core/dataprocessing.py | 5 ++- src/damast/core/decorators.py | 22 +++++++++++- src/damast/core/metadata.py | 35 +++++++++++++------ src/damast/core/polars_dataframe.py | 13 ++++++- src/damast/core/transformations.py | 14 ++++++++ tests/damast/cli/test_cli.py | 18 ++++++---- tests/damast/core/test_dataframe.py | 26 ++++++++++++-- tests/damast/core/test_dataprocessing.py | 30 ++++++++++++++-- .../domains/maritime/ais/test_augmenters.py | 12 +++++-- 12 files changed, 169 insertions(+), 34 deletions(-) diff --git a/src/damast/cli/data_annotate.py b/src/damast/cli/data_annotate.py index 018d452..c6334ee 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_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) diff --git a/src/damast/cli/data_inspect.py b/src/damast/cli/data_inspect.py index bf0e2c5..2cfb5e9 100644 --- a/src/damast/cli/data_inspect.py +++ b/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 = "" diff --git a/src/damast/core/dataframe.py b/src/damast/core/dataframe.py index d495601..79646d5 100644 --- a/src/damast/core/dataframe.py +++ b/src/damast/core/dataframe.py @@ -216,6 +216,9 @@ def from_files(cls, 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 @@ -341,6 +344,7 @@ 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: """ @@ -349,7 +353,10 @@ def from_file(cls, :param filename: Filename to use for importing and creating the annotated dataframe """ - return cls.from_files([filename], metadata_required=metadata_required, merge_strategy=merge_strategy) + 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: @@ -447,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) diff --git a/src/damast/core/dataprocessing.py b/src/damast/core/dataprocessing.py index 24866b4..459734d 100644 --- a/src/damast/core/dataprocessing.py +++ b/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 } diff --git a/src/damast/core/decorators.py b/src/damast/core/decorators.py index e1ad8c9..299b021 100644 --- a/src/damast/core/decorators.py +++ b/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( diff --git a/src/damast/core/metadata.py b/src/damast/core/metadata.py index 0c3c7a6..91838e9 100644 --- a/src/damast/core/metadata.py +++ b/src/damast/core/metadata.py @@ -578,9 +578,9 @@ def apply( if validation_mode == ValidationMode.IGNORE: return df + xdf = XDataFrame(df) # Check if representation type is the same and apply known metadata if validation_mode == ValidationMode.READONLY: - xdf = XDataFrame(df) if self.representation_type is not None: dtype = xdf.dtype(column_name) if dtype.from_python(self.representation_type) != self.representation_type and \ @@ -606,7 +606,6 @@ def apply( return df if validation_mode == ValidationMode.UPDATE_DATA: - xdf = XDataFrame(df) if self.representation_type is not None: dtype = xdf.dtype(column_name) if self.representation_type not in [dtype, dtype.to_python()]: @@ -639,6 +638,8 @@ def apply( return xdf._dataframe if validation_mode == ValidationMode.UPDATE_METADATA: + if self.representation_type is None: + self.representation_type = xdf.dtype(column_name) return self.update_datarange_and_stats(df, column_name) def update_datarange_and_stats(self, df, column_name: str): @@ -778,6 +779,9 @@ def merge(self, other: DataSpecification, strategy: MergeStrategy | None = None) :raises ValueError: If the data-specifications have overlapping attributes, that have distinct non-``None`` values, the function throws an error. """ + if type(other) is not DataSpecification: + raise TypeError(f"{self.__class__.__name__}.merge: cannot merge self of type={type(self)} with other of type={type(other)}") + if self.name != other.name: raise ValueError( f"{self.__class__.__name__}.merge: cannot merge specs with different name property" @@ -810,16 +814,17 @@ def merge(self, other: DataSpecification, strategy: MergeStrategy | None = None) merged_value = this_value.merge(other_value) setattr(ds, key.value, merged_value) return ds - except Exception as e: - if not strategy: - raise ValueError( - f"{self.__class__.__name__}.merge cannot merge specs for '{self.name}': value for '{key.value}' differs: " + except Exception: + logger.debug("Merge failed: {e}") - f" on self: '{this_value}' vs. other: '{other_value}' -- {e}" - ) + if not strategy: + raise ValueError( + f"{self.__class__.__name__}.merge cannot merge specs for '{self.name}': value for '{key.value}' differs: " - logger.info(f"{self.__class__.__name__}.merge: using merge strategy {strategy} for {key.value}: this={this_value} -- other={other_value}") + f" on self: '{this_value}' vs. other: '{other_value}'" + ) + logger.info(f"{self.__class__.__name__}.merge: using merge strategy {strategy} for {key.value}: this={this_value} -- other={other_value}") if strategy == DataSpecification.MergeStrategy.OTHER: setattr(ds, key.value, other_value) elif strategy == DataSpecification.MergeStrategy.THIS: @@ -842,6 +847,13 @@ def merge_lists( :param a_specs: First list of specs :param b_specs: Second list of specs """ + + if type(a_specs) is not list: + raise TypeError(f"{cls.__name__}.merge_lists: cannot merge {type(a_specs)} -- needs to be list(DataSpecificiation)") + + if type(b_specs) is not list: + raise TypeError(f"{cls.__name__}.merge_lists: cannot merge {type(a_specs)} -- needs to be list(DataSpecificiation)") + result_specs: List[DataSpecification] = [] b_column_dict = {x.name: x for x in b_specs} @@ -1199,7 +1211,7 @@ def apply( delta = set(columns).difference([x.name for x in self.columns]) if delta: raise ValueError( - f"{self.__class__.__name__}.apply: missing column metatadata for columns: {','.join(delta)}. " + f"{self.__class__.__name__}.apply: missing column metadata for columns: {','.join(delta)}. " f"Column data exists in dataframe, but no metadata is available" ) @@ -1293,6 +1305,9 @@ def specfile(cls, files: list[str | Path]) -> str: return Path(commonpath) / f"{commonprefix}.collection{DAMAST_SPEC_SUFFIX}" def merge(self, other: MetaData, strategy: DataSpecification.MergeStrategy | None = None) -> MetaData: + if type(other) is not MetaData: + raise TypeError(f"{self.__class__.__name__}.merge: cannot merge MetaData with other of type={type(other)}") + column_specs = DataSpecification.merge_lists(self.columns, other.columns, strategy) annotations = [] for k,v in self.annotations.items(): diff --git a/src/damast/core/polars_dataframe.py b/src/damast/core/polars_dataframe.py index 5dce0d7..1e6b63f 100644 --- a/src/damast/core/polars_dataframe.py +++ b/src/damast/core/polars_dataframe.py @@ -407,7 +407,16 @@ def import_hdf5(cls, files: str | Path | list[str|Path]) -> tuple[polars.LazyFra data_frames = [] for filename in files: - data_frames.append( pandas.read_hdf(str(filename)) ) + pandas_df = pandas.read_hdf(str(filename)) + + with tables.open_file(str(filename)) as hdf5file: + for column in pandas_df.columns: + column_attrs = hdf5file.get_node(f"/dataframe/columns/{column}")._v_attrs + if "representation_type" in column_attrs: + if column_attrs["representation_type"] == "int": + pandas_df[column] = pandas_df[column].astype("Int64") + + data_frames.append(pandas_df) pandas_df = pandas.concat(data_frames, ignore_index=True) df = polars.from_pandas(pandas_df) @@ -440,6 +449,8 @@ def export_hdf5(cls, df: polars.DataFrame | polars.LazyFrame, path: str | Path) if isinstance(df, polars.dataframe.DataFrame): df = df.lazy() + # An export of an int column with NaNs will automatically convert this to Float64 + # while Int64 could be used, the underlying exporter pytables does not (yet) support this pandas_df = df.collect().to_pandas() pandas_df.to_hdf(path, key=DAMAST_HDF5_ROOT) return path diff --git a/src/damast/core/transformations.py b/src/damast/core/transformations.py index 2b13324..db2d94a 100644 --- a/src/damast/core/transformations.py +++ b/src/damast/core/transformations.py @@ -136,6 +136,20 @@ def _get_name(self, name: str, datasource: str | None) -> any: resolved_name = name_mappings[resolved_name] name = name.replace(match.group(), resolved_name) + + # If multiple sources are involved, allow to use the pattern {{:}} + m = re.match("{{(\\w+):(\\w+)}}", name) + if m: + source_label = m.groups()[0] + field_name = m.groups()[1] + if source_label not in self.name_mappings: + raise RuntimeError(f"{self.__class__.__name__}.get_name: unknown source '{source_label}' defined in {name}") + + resolved_name = field_name + if field_name in self.name_mappings[source_label]: + resolved_name = self.name_mappings[source_label][field_name] + name = name.replace(m.group(), resolved_name) + return name @abstractmethod diff --git a/tests/damast/cli/test_cli.py b/tests/damast/cli/test_cli.py index 8ed183c..dac08fc 100644 --- a/tests/damast/cli/test_cli.py +++ b/tests/damast/cli/test_cli.py @@ -59,16 +59,20 @@ def test_subparser(name, klass, script_runner): for option in a.option_strings: assert re.search(option, result.stdout) is not None, f"Should have {option=}" -@pytest.mark.parametrize("filename", [ - "data.hdf5", - "test_ais.parquet", - "test_ais.csv", - "test_dataframe.csv", - "test_dataframe.hdf5", +@pytest.mark.parametrize("filename, incomplete_metadata", [ + ["data.hdf5", True], + ["test_ais.parquet", False], + ["test_ais.csv", False], + ["test_dataframe.csv", False], + ["test_dataframe.hdf5", True], ]) -def test_inspect(data_path, filename, script_runner): +def test_inspect(data_path, filename, incomplete_metadata, script_runner): result = script_runner.run(['damast', 'inspect', '-f', str(data_path / filename)]) + if incomplete_metadata: + assert result.returncode != 0 + result = script_runner.run(['damast', 'inspect', '-f', str(data_path / filename), '--validation-mode', 'ignore']) + assert re.search(r"Loading dataframe \(1 files\)", result.stdout) is not None, "Process dataframe" assert re.search("shape:", result.stdout) is not None, "Show dataframe" diff --git a/tests/damast/core/test_dataframe.py b/tests/damast/core/test_dataframe.py index eedebfd..3fe2fb2 100644 --- a/tests/damast/core/test_dataframe.py +++ b/tests/damast/core/test_dataframe.py @@ -22,7 +22,7 @@ @pytest.fixture() def metadata(): - column_spec = DataSpecification(name="height", + column_spec_height = DataSpecification(name="height", category=DataCategory.STATIC, unit=units.m, abbreviation="height", @@ -32,7 +32,12 @@ def metadata(): comment = Annotation(name=Annotation.Key.Comment, value="test dataframe") annotations = [license, comment] - return MetaData(columns=[column_spec], annotations=annotations) + column_spec_letter = DataSpecification(name="letter", + category=DataCategory.STATIC, + abbreviation="letter") + + annotations = [license, comment] + return MetaData(columns=[column_spec_height, column_spec_letter], annotations=annotations) @pytest.fixture() @@ -77,6 +82,18 @@ def test_annotated_dataframe_deep_copy(metadata, polars_dataframe): assert adf.dataframe.column_names == column_names assert adf_copy.dataframe.column_names != column_names +def test_incomplete_metadata(metadata, polars_dataframe, tmp_path): + """ + Simple test of the annotated dataframe export to HDF5 + + :param metadata: metadata to use + :param polars_dataframe: polars dataframe to use + :param tmp_path: where to temporarily save the data to HDF5 + """ + incomplete_metadata = MetaData(columns=metadata.columns[:-1]) + with pytest.raises(ValueError, match=f"missing column metadata for columns: {metadata.columns[-1].name}"): + AnnotatedDataFrame(dataframe=polars_dataframe, + metadata=incomplete_metadata) def test_annotated_dataframe_export_hdf5(metadata, polars_dataframe, tmp_path): """ @@ -147,7 +164,10 @@ def test_annotated_dataframe_import_vaex_hdf5(data_path): """ hdf5_path = data_path / "data.hdf5" - adf = AnnotatedDataFrame.from_file(hdf5_path) + with pytest.raises(ValueError, match="missing column metadata"): + AnnotatedDataFrame.from_file(hdf5_path) + + adf = AnnotatedDataFrame.from_file(hdf5_path, metadata_required=False, validation_mode=ValidationMode.IGNORE) assert adf.column_names == ["height", "letter"] pandas_df = pd.DataFrame(data = {'height': [0,1,2], 'letter': ['a','b','c']}) diff --git a/tests/damast/core/test_dataprocessing.py b/tests/damast/core/test_dataprocessing.py index 103ec3c..1aae8ed 100644 --- a/tests/damast/core/test_dataprocessing.py +++ b/tests/damast/core/test_dataprocessing.py @@ -150,12 +150,19 @@ def __init__(self): "lon": {} }, label='other' ) - @damast.core.output({}) + @damast.core.output({"{{other:event_type}}": { 'description': 'hozint event type', 'representation_type': str}, + "{{other:lat}}": {}, + "{{other:lon}}": {}, + }) def transform(self, df: AnnotatedDataFrame, other: AnnotatedDataFrame) -> AnnotatedDataFrame: 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) + + updated_metadata = df._metadata.merge(other._metadata) + updated_metadata.drop(other_timestamp) + df._metadata = updated_metadata return df class JoinSpatioTemporal(PipelineElement): @@ -190,6 +197,8 @@ def __init__(self, 'event_type': {}, 'event_delta_distance': { 'description': "Distance between vessel and event", 'unit': 'km'}, 'event_delta_time': { 'description': "Timedelta between event and vessel message", 'unit': 's'}, + 'latitude': {'description': 'event latitude', 'unit': 'deg' }, + 'longitude': {'description': 'event longitude', 'unit': 'deg' }, }) def transform(self, df: AnnotatedDataFrame, other: AnnotatedDataFrame) -> AnnotatedDataFrame: import polars as pl @@ -221,6 +230,10 @@ def transform(self, df: AnnotatedDataFrame, other: AnnotatedDataFrame) -> Annota ), event_delta_time = pl.col(other_timestamp) - pl.col(df_timestamp) ) + + updated_metadata = df._metadata.merge(other._metadata) + updated_metadata.drop(other_timestamp) + df._metadata = updated_metadata return df @@ -233,9 +246,20 @@ def height_metadata(): metadata = MetaData(columns=[column_spec]) return metadata - @pytest.fixture() def height_dataframe(): + data = [ + [0], + [1], + [2] + ] + columns = [ + "height" + ] + return polars.LazyFrame(data, columns, orient="row") + +@pytest.fixture() +def height_letter_dataframe(): data = [ [0, "a"], [1, "b"], @@ -442,7 +466,7 @@ def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame: assert "status_suffix" in adf.column_names, "Expect 'status_suffix' to be a new column" assert adf.metadata['status_suffix'], "Expect metadata to be available for 'status_suffix'" - assert adf.metadata['status_suffix'].representation_type == polars.Int64, "Expect representation_type Int64 for 'status_suffix'" + assert adf.metadata['status_suffix'].representation_type == polars.Int64, f"Expect representation_type Int64 for 'status_suffix', but got {adf.metadata['status_suffix']}" @pytest.mark.parametrize("varname",["x","xyz"]) def test_decorator_renaming(varname, tmp_path): diff --git a/tests/damast/domains/maritime/ais/test_augmenters.py b/tests/damast/domains/maritime/ais/test_augmenters.py index 9bdf8bc..8ad550d 100644 --- a/tests/damast/domains/maritime/ais/test_augmenters.py +++ b/tests/damast/domains/maritime/ais/test_augmenters.py @@ -60,7 +60,11 @@ def test_add_missing_ais_status(tmp_path): # Create annotated dataframe adf = damast.core.AnnotatedDataFrame( df, - damast.core.MetaData(columns=[damast.core.DataSpecification("Status", representation_type=int)]) + damast.core.MetaData(columns=[ + damast.core.DataSpecification("Status", representation_type=int), + damast.core.DataSpecification("MMSI", representation_type=int), + damast.core.DataSpecification("timestamp", representation_type=pl.Datetime) + ]) ) # Create pipeline @@ -324,7 +328,11 @@ def test_message_index(tmp_path): columns=[damast.core.DataSpecification(ColumnName.MMSI, representation_type=int), damast.core.DataSpecification(ColumnName.TIMESTAMP, representation_type=pl.datatypes.Datetime("ns")), damast.core.DataSpecification(ColumnName.LATITUDE, unit=units.deg, representation_type=pl.Float64), - damast.core.DataSpecification(ColumnName.LONGITUDE, unit=units.deg, representation_type=pl.Float64)]) + damast.core.DataSpecification(ColumnName.LONGITUDE, unit=units.deg, representation_type=pl.Float64), + damast.core.DataSpecification("REF INDEX", representation_type=pl.Int64), + damast.core.DataSpecification("INVERSE REF", representation_type=pl.Int64), + ] + ) adf = damast.core.AnnotatedDataFrame(df, metadata) # Create pipeline From e234d231a9e95d6cc6e4be9bb52e3bd9a63104cb Mon Sep 17 00:00:00 2001 From: Thomas Roehr Date: Tue, 7 Jul 2026 12:37:25 +0200 Subject: [PATCH 3/6] core: embed merging of metadata in join transformer --- src/damast/core/dataframe.py | 2 +- src/damast/core/metadata.py | 6 ++++-- tests/damast/core/test_dataprocessing.py | 8 ++------ 3 files changed, 7 insertions(+), 9 deletions(-) diff --git a/src/damast/core/dataframe.py b/src/damast/core/dataframe.py index 79646d5..3e52bd8 100644 --- a/src/damast/core/dataframe.py +++ b/src/damast/core/dataframe.py @@ -443,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): diff --git a/src/damast/core/metadata.py b/src/damast/core/metadata.py index 91838e9..505811e 100644 --- a/src/damast/core/metadata.py +++ b/src/damast/core/metadata.py @@ -1221,12 +1221,14 @@ def __contains__(self, column_name: str): """""" return any(colum_spec.name == column_name for colum_spec in self.columns) - def drop(self, columns: str | list[str]): + def drop(self, columns: str | list[str]) -> MetaData: """ Drop specification by column name """ columns = [columns] if type(columns) is str else columns - self.columns = [x for x in self.columns if x.name not in columns] + updated_columns = [x for x in self.columns if x.name not in columns] + + return MetaData(updated_columns, [y for _,y in self.annotations.items()]) def __getitem__(self, column_name: str) -> DataSpecification: """ diff --git a/tests/damast/core/test_dataprocessing.py b/tests/damast/core/test_dataprocessing.py index 1aae8ed..1587a0d 100644 --- a/tests/damast/core/test_dataprocessing.py +++ b/tests/damast/core/test_dataprocessing.py @@ -160,9 +160,7 @@ def transform(self, df: AnnotatedDataFrame, other: AnnotatedDataFrame) -> Annota df._dataframe = df.join(other._dataframe, left_on=df_timestamp, right_on=other_timestamp) - updated_metadata = df._metadata.merge(other._metadata) - updated_metadata.drop(other_timestamp) - df._metadata = updated_metadata + df._metadata = df._metadata.merge(other._metadata).drop(other_timestamp) return df class JoinSpatioTemporal(PipelineElement): @@ -231,9 +229,7 @@ def transform(self, df: AnnotatedDataFrame, other: AnnotatedDataFrame) -> Annota event_delta_time = pl.col(other_timestamp) - pl.col(df_timestamp) ) - updated_metadata = df._metadata.merge(other._metadata) - updated_metadata.drop(other_timestamp) - df._metadata = updated_metadata + df._metadata = df._metadata.merge(other._metadata).drop(other_timestamp) return df From 753877674c41d189e84d7f8b65e6e64717260064 Mon Sep 17 00:00:00 2001 From: Thomas Roehr Date: Tue, 7 Jul 2026 13:40:02 +0200 Subject: [PATCH 4/6] core: validate metadata after each pipeline steps --- src/damast/core/decorators.py | 1 + src/damast/data_handling/transformers/augmenters.py | 4 ++-- src/damast/domains/maritime/transformers/features.py | 2 +- tests/damast/domains/maritime/test_data_processing.py | 3 +++ 4 files changed, 7 insertions(+), 3 deletions(-) diff --git a/src/damast/core/decorators.py b/src/damast/core/decorators.py index 299b021..5b08a91 100644 --- a/src/damast/core/decorators.py +++ b/src/damast/core/decorators.py @@ -200,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: diff --git a/src/damast/data_handling/transformers/augmenters.py b/src/damast/data_handling/transformers/augmenters.py index 7f664a5..7db306d 100644 --- a/src/damast/data_handling/transformers/augmenters.py +++ b/src/damast/data_handling/transformers/augmenters.py @@ -262,13 +262,13 @@ def transform(self, df: damast.core.AnnotatedDataFrame) -> damast.core.Annotated dataframe = dataframe\ .sort(group_column, time_column)\ .with_columns( - pl.from_epoch(pl.col(time_column), time_unit="s").diff().dt.total_seconds().over(group_column).alias('delta_time') + pl.from_epoch(pl.col(time_column), time_unit="s").diff().dt.total_seconds(fractional=True).over(group_column).alias('delta_time') ) else: dataframe = dataframe\ .sort(group_column, time_column)\ .with_columns( - pl.col(time_column).diff().dt.total_seconds().over(group_column).alias('delta_time') + pl.col(time_column).diff().dt.total_seconds(fractional=True).over(group_column).alias('delta_time') ) dataframe = dataframe\ diff --git a/src/damast/domains/maritime/transformers/features.py b/src/damast/domains/maritime/transformers/features.py index 5335711..a0b7e14 100644 --- a/src/damast/domains/maritime/transformers/features.py +++ b/src/damast/domains/maritime/transformers/features.py @@ -177,7 +177,7 @@ def transform(self, dataframe = dataframe.sort(group, sort_column).with_columns( pl.col(heading).diff().alias(delta_heading), - pl.col(sort_column).diff().dt.total_seconds().alias("_delta_time") + pl.col(sort_column).diff().dt.total_seconds(fractional=True).alias("_delta_time") ) dataframe = dataframe.with_columns( diff --git a/tests/damast/domains/maritime/test_data_processing.py b/tests/damast/domains/maritime/test_data_processing.py index d14869f..ebee4e4 100644 --- a/tests/damast/domains/maritime/test_data_processing.py +++ b/tests/damast/domains/maritime/test_data_processing.py @@ -195,6 +195,9 @@ def test_aggregate(data_path, tmp_path): assert df.metadata['lat'].unit == 'deg' assert df.metadata['lon'].unit == 'deg' + assert df.metadata['delta_distance'].representation_type == polars.Float64 + assert df.compat.dtype('delta_distance') == polars.Float64 + assert "delta_distance" in df.column_names def test_aggregate_with_name_mapping(data_path, tmp_path): From 9f7d81ccedd6cda3f96e9be80145cb5806e29d43 Mon Sep 17 00:00:00 2001 From: Thomas Roehr Date: Mon, 13 Jul 2026 15:36:39 +0200 Subject: [PATCH 5/6] Update logging --- src/damast/cli/main.py | 25 +++++++++++++++++++++++-- src/damast/config.py | 3 +++ 2 files changed, 26 insertions(+), 2 deletions(-) create mode 100644 src/damast/config.py diff --git a/src/damast/cli/main.py b/src/damast/cli/main.py index be7277e..7b508bb 100644 --- a/src/damast/cli/main.py +++ b/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) diff --git a/src/damast/config.py b/src/damast/config.py new file mode 100644 index 0000000..8c2297d --- /dev/null +++ b/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' From 2ed502010077388ebb087adceab2c10698d9ea4d Mon Sep 17 00:00:00 2001 From: Thomas Roehr Date: Mon, 13 Jul 2026 15:38:30 +0200 Subject: [PATCH 6/6] core: update datetime handling --- src/damast/core/data_description.py | 28 +++++++++++++++++++--------- src/damast/core/metadata.py | 25 ++++++++++++------------- src/damast/core/polars_dataframe.py | 28 ++++++++++++++++++++++++++-- tests/damast/core/test_metadata.py | 7 +++++++ 4 files changed, 64 insertions(+), 24 deletions(-) diff --git a/src/damast/core/data_description.py b/src/damast/core/data_description.py index 772c243..5df5303 100644 --- a/src/damast/core/data_description.py +++ b/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): @@ -201,7 +207,7 @@ 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 ListOfValues: + 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)) @@ -263,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 diff --git a/src/damast/core/metadata.py b/src/damast/core/metadata.py index 505811e..c68f9b2 100644 --- a/src/damast/core/metadata.py +++ b/src/damast/core/metadata.py @@ -654,13 +654,12 @@ def update_datarange_and_stats(self, df, column_name: str): min_value, max_value = df.compat.minmax(column_name) if min_value is not None and max_value is not None: self.value_range = MinMax(min_value, max_value) - elif df.compat.is_numeric(column_name): + elif df.compat.is_numeric(column_name) or df.compat.is_datetime(column_name): try: min_value, max_value = df.compat.minmax(column_name) if min_value is not None and max_value is not None: logger.debug(f"Setting value range: MinMax for {column_name}") self.value_range = MinMax(min_value, max_value) - results = df.compat.minmax_stats([column_name]) logger.debug(f"Setting value stats for {column_name}") @@ -796,26 +795,26 @@ def merge(self, other: DataSpecification, strategy: MergeStrategy | None = None) this_value = getattr(self, key.value) other_value = getattr(other, key.value) - if key == self.Key.representation_type: - if hasattr(this_value, "to_python"): - this_value = this_value.to_python() - if hasattr(other_value, "to_python"): - other_value = other_value.to_python() - - if this_value is None: + if this_value == other_value: + setattr(ds, key.value, this_value) + elif this_value is None: setattr(ds, key.value, other_value) elif other_value is None: setattr(ds, key.value, this_value) - elif this_value == other_value: - setattr(ds, key.value, this_value) else: + if key == self.Key.representation_type: + if hasattr(this_value, "to_python"): + this_value = this_value.to_python() + if hasattr(other_value, "to_python"): + other_value = other_value.to_python() + try: if hasattr(this_value, "merge"): merged_value = this_value.merge(other_value) setattr(ds, key.value, merged_value) return ds - except Exception: - logger.debug("Merge failed: {e}") + except Exception as e: + logger.warning(f"Merge failed: {e}") if not strategy: raise ValueError( diff --git a/src/damast/core/polars_dataframe.py b/src/damast/core/polars_dataframe.py index 1e6b63f..04a668e 100644 --- a/src/damast/core/polars_dataframe.py +++ b/src/damast/core/polars_dataframe.py @@ -61,6 +61,15 @@ def types(cls) -> dict[str, any]: @classmethod def resolve_type(cls, type_txt: str): + if type_txt == "datetime": + type_txt = "Datetime" + elif type_txt == "str": + type_txt = "String" + elif type_txt == "int": + type_txt = "Int64" + elif type_txt == "float": + type_txt = "Float64" + return eval(type_txt, cls.types()) @@ -92,6 +101,12 @@ def is_string(self, column_name: str) -> bool: def is_numeric(self, column_name: str) -> bool: return self.dtype(column_name).is_numeric() + def is_datetime(self, column_name: str) -> bool: + return type(self.dtype(column_name)) is polars.Datetime + + def is_date(self, column_name: str) -> bool: + return type(self.dtype(column_name)) is polars.Date + def __getitem__(self, column_name: str): """ Make dataframe subscriptable and behave more like the :class:`pandas.DataFrame`. @@ -194,12 +209,21 @@ def minmax_stats(self, column_names: list[str]) -> dict[str, dict[str, any]]: fields.extend([ polars.col(column).min().alias(f"{column}_min_value"), polars.col(column).max().alias(f"{column}_max_value"), - polars.col(column).mean().alias(f"{column}_mean"), - polars.col(column).std().alias(f"{column}_stddev"), polars.col(column).count().alias(f"{column}_total_count"), polars.col(column).null_count().alias(f"{column}_null_count") ]) + if self.is_datetime(column): + fields.extend([ + (polars.col(column).dt.timestamp("us").mean() / 1_000_000).alias(f"{column}_mean"), + (polars.col(column).dt.timestamp("us").std() / 1_000_000).alias(f"{column}_stddev") + ]) + else: + fields.extend([ + polars.col(column).mean().alias(f"{column}_mean"), + polars.col(column).std().alias(f"{column}_stddev"), + ]) + result = self._dataframe.select( fields ).collect() diff --git a/tests/damast/core/test_metadata.py b/tests/damast/core/test_metadata.py index c729c72..5d9104d 100644 --- a/tests/damast/core/test_metadata.py +++ b/tests/damast/core/test_metadata.py @@ -154,6 +154,13 @@ def test_data_specification_read_write(name, category, is_optional, category=DataCategory.DYNAMIC, representation_type=int), DataSpecification.MergeStrategy.THIS, None + ], + [DataSpecification(name="a", + representation_type=float, + unit='deg'), + DataSpecification(name="a", + unit='rad'), + DataSpecification.MergeStrategy.OTHER, None ] ]) def test_data_specification_merge(dataspec, other_dataspec, merge_strategy, error_msg):