add time and date type transformations

Author: mathleur

polytope_feature/datacube/transformations/datacube_type_change/datacube_type_change.py

@@ -1,5 +1,6 @@
 from copy import deepcopy
 from importlib import import_module
+import pandas as pd
 
 from ..datacube_transformations import DatacubeAxisTransformation
 
@@ -75,4 +76,46 @@ def make_str(self, value):
         return tuple(values)
 
 
-_type_to_datacube_type_change_lookup = {"int": "TypeChangeStrToInt"}
+class TypeChangeStrToTimestamp(DatacubeAxisTypeChange):
+    def __init__(self, axis_name, new_type):
+        self.axis_name = axis_name
+        self._new_type = new_type
+
+    def transform_type(self, value):
+        try:
+            return pd.Timestamp(value)
+        except ValueError:
+            return None
+
+    def make_str(self, value):
+        values = []
+        for val in value:
+            values.append(val.strftime('%Y%m%d'))
+        return tuple(values)
+
+
+class TypeChangeStrToTimedelta(DatacubeAxisTypeChange):
+    def __init__(self, axis_name, new_type):
+        self.axis_name = axis_name
+        self._new_type = new_type
+
+    def transform_type(self, value):
+        try:
+            hours = int(value[:2])
+            mins = int(value[2:])
+            return pd.Timedelta(hours=hours, minutes=mins)
+        except ValueError:
+            return None
+
+    def make_str(self, value):
+        values = []
+        for val in value:
+            hours = int(val.total_seconds() // 3600)
+            mins = int((val.total_seconds() % 3600) // 60)
+            values.append(f"{hours:02d}{mins:02d}")
+        return tuple(values)
+
+
+_type_to_datacube_type_change_lookup = {"int": "TypeChangeStrToInt",
+                                        "date": "TypeChangeStrToTimestamp",
+                                        "time": "TypeChangeStrToTimedelta"}

tests/test_date_time_unmerged.py

@@ -0,0 +1,67 @@
+import pandas as pd
+import pytest
+
+from polytope_feature.polytope import Polytope, Request
+from polytope_feature.shapes import Box, Select
+
+
+class TestSlicingFDBDatacube:
+    def setup_method(self, method):
+        # Create a dataarray with 3 labelled axes using different index types
+        self.options = {
+            "axis_config": [
+                {"axis_name": "step", "transformations": [{"name": "type_change", "type": "int"}]},
+                {"axis_name": "date", "transformations": [{"name": "type_change", "type": "date"}]},
+                {"axis_name": "time", "transformations": [{"name": "type_change", "type": "time"}]},
+                {
+                    "axis_name": "values",
+                    "transformations": [
+                        {"name": "mapper", "type": "octahedral", "resolution": 1280, "axes": ["latitude", "longitude"]}
+                    ],
+                },
+                {"axis_name": "latitude", "transformations": [{"name": "reverse", "is_reverse": True}]},
+                {"axis_name": "longitude", "transformations": [{"name": "cyclic", "range": [0, 360]}]},
+            ],
+            "pre_path": {"class": "od", "expver": "0001", "levtype": "sfc", "stream": "oper", "type": "fc"},
+            "compressed_axes_config": [
+                "longitude",
+                "latitude",
+                "levtype",
+                "step",
+                "date",
+                "domain",
+                "expver",
+                "param",
+                "class",
+                "stream",
+                "type",
+            ],
+        }
+
+    # Testing different shapes
+    @pytest.mark.fdb
+    def test_fdb_datacube(self):
+        import pygribjump as gj
+
+        request = Request(
+            Select("step", [0]),
+            Select("levtype", ["sfc"]),
+            Select("date", [pd.Timestamp("20240118")]),
+            Select("time", [pd.Timedelta("00:00:00")]),
+            Select("domain", ["g"]),
+            Select("expver", ["0001"]),
+            Select("param", ["167"]),
+            Select("class", ["od"]),
+            Select("stream", ["oper"]),
+            Select("type", ["fc"]),
+            Box(["latitude", "longitude"], [0, 0], [0.2, 0.2]),
+        )
+        self.fdbdatacube = gj.GribJump()
+        self.API = Polytope(
+            datacube=self.fdbdatacube,
+            options=self.options,
+        )
+        result = self.API.retrieve(request)
+        result.pprint()
+        assert len(result.leaves) == 3
+        assert len(result.leaves[0].result) == 3