core: update datetime handling

Author: 2maz

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

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(

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()

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):