Added basic dataprocessor.

glotaran/io/preprocssor/preprocessor.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+import abc
+from typing import Annotated
+from typing import Literal
+
+try:
+    from typing import Self
+except ImportError:
+    Self = "PreProcessor"
+
+import xarray as xr
+from pydantic import BaseModel
+from pydantic import Field
+
+
+class PreProcessorAction(BaseModel, abc.ABC):
+    class Config:
+        arbitrary_types_allowed = True
+
+    @abc.abstractmethod
+    def apply(self, data: xr.DataArray) -> xr.DataArray:
+        pass
+
+
+class CorrectBaselineValue(PreProcessorAction):
+    action: Literal["baseline-value"] = "baseline-value"
+    value: float
+
+    def apply(self, data: xr.DataArray) -> xr.DataArray:
+        return data - self.value
+
+
+class CorrectBaselineAverage(PreProcessorAction):
+    action: Literal["baseline-average"] = "baseline-average"
+    average: dict[str, slice]
+
+    def apply(self, data: xr.DataArray) -> xr.DataArray:
+        selection = data
+        for axis, interval in self.average.items():
+            selection = selection.sel({axis: interval})
+        return data - (selection.sum() / selection.size)
+
+
+# TODO
+#  class ShiftAlongAxis(PreProcessorAction):
+#      action: Literal["shift-along-axis"] = "shift-along-axis"
+#      axis: str
+#      value: float | list[float]
+
+
+# TODO
+#  class AverageAxis(PreProcessorAction):
+#      action: Literal["average"] = "average"
+#      new_coord: Literal["first", "last", "mean"] = "mean"
+#      axis: str
+#      number: int
+
+
+class PreProcessorActions(BaseModel):
+    __root__: Annotated[
+        CorrectBaselineValue | CorrectBaselineAverage,
+        Field(discriminator="action"),
+    ]
+
+
+class PreProcessor:
+    def __init__(self, original: xr.DataArray, actions: list[PreProcessorAction] | None = None):
+        self._actions: list[PreProcessorAction] = actions or []
+        self._original: xr.DataArray = original
+
+    def get_result(self) -> xr.DataArray:
+        data = self._original
+        for action in self._actions:
+            data = action.apply(data)
+        return data
+
+    def push_action(self, action: PreProcessorAction):
+        self._actions.append(action)
+
+    def correct_baseline_value(self, value: float) -> Self:
+        self.push_action(CorrectBaselineValue(value=value))
+        return self