Updating the ProcessStep definition (#34)

* Updated ProcessStep definition
* Finished tests for ProcessingStep

Author: malte-storm

src/modacor/dataclasses/basedata.py

@@ -1,3 +1,5 @@
+__all__ = ["BaseData"]
+
 # import tiled
 # import tiled.client
 import logging
@@ -34,7 +36,7 @@ class BaseData:
     """
 
     # Core data array stored as an xarray DataArray
-    signal: np.ndarray = field(factory=np.ndarray, validator=[v.instance_of(np.ndarray)])
+    signal: np.ndarray = field(default=np.array(()), validator=[v.instance_of(np.ndarray)])
 
     # Dict of variances represented as xarray DataArray objects; defaulting to an empty dict
     variances: Dict[str, np.ndarray] = field(factory=dict, validator=[v.instance_of(dict)])

src/modacor/dataclasses/process_step.py

@@ -1,4 +1,26 @@
 # SPDX-License-Identifier: BSD-3-Clause
+# Copyright 2025 MoDaCor Authors
+#
+# Redistribution and use in source and binary forms, with or without modification,
+# are permitted provided that the following conditions are met:
+# 1. Redistributions of source code must retain the above copyright notice, this
+#    list of conditions and the following disclaimer.
+# 2. Redistributions in binary form must reproduce the above copyright notice,
+#    this list of conditions and the following disclaimer in the documentation
+#    and/or other materials provided with the distribution.
+# 3. Neither the name of the copyright holder nor the names of its contributors
+#    may be used to endorse or promote products derived from this software without
+#    specific prior written permission.
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS “AS IS” AND
+# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR
+# ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
+# ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 
 __all__ = ["ProcessStep"]
@@ -9,7 +31,7 @@
 from abc import abstractmethod
 from numbers import Integral
 from pathlib import Path
-from typing import Any
+from typing import Any, Iterable, Type
 
 from attrs import define, field
 from attrs import validators as v
@@ -18,13 +40,31 @@
 from .databundle import DataBundle
 from .messagehandler import MessageHandler
 from .process_step_describer import ProcessStepDescriber
+from .processing_data import ProcessingData
 from .validators import is_list_of_ints
 
 
 @define
 class ProcessStep:
     """A base class defining a processing step"""
 
+    # Class attributes for the process step
+    CONFIG_KEYS = {
+        "with_processing_keys": {
+            "type": str,
+            "allow_iterable": True,
+            "allow_none": True,
+            "default": None,
+        },
+        "output_processing_key": {
+            "type": str,
+            "allow_iterable": False,
+            "allow_none": True,
+            "default": None,
+        },
+    }
+
+    # The configuration keys for the process step instantiation
     io_sources: IoSources = field()
 
     # class attribute for a machine-readable description of the process step
@@ -36,7 +76,10 @@ class ProcessStep:
     )
 
     # dynamic instance configuration
-    configuration: dict = field(factory=dict, validator=v.instance_of(dict))
+    configuration: dict = field(
+        factory=dict,
+        validator=lambda inst, attrs, val: inst.is_process_step_dict,
+    )
 
     # flags and attributes for running the pipeline
     requires_steps: list[int] = field(factory=list, validator=is_list_of_ints)
@@ -54,6 +97,15 @@ class ProcessStep:
 
     # internal variables:
     __prepared: bool = field(default=False, validator=v.instance_of(bool))
+    processing_data: ProcessingData = field(
+        default=None, validator=v.optional(v.instance_of(ProcessingData))
+    )
+
+    def __attrs_post_init__(self):
+        """
+        Post-initialization method to set up the process step.
+        """
+        self.configuration = self.default_config()
 
     def prepare_execution(self):
         """
@@ -64,30 +116,24 @@ def prepare_execution(self):
         """
         pass
 
-    def can_execute(self, input_field_names: list[str]) -> bool:
-        """
-        Check if the process step can be executed
-
-        The default implementation always returns True and any ProcessStep
-        that has dependency checks should override this method.
-        """
-        return True
-
     @abstractmethod
-    def calculate(self, data: DataBundle, **kwargs: Any) -> dict[str, Any]:
+    def calculate(self) -> dict[str, DataBundle]:
         """Calculate the process step on the given data"""
         raise NotImplementedError("Subclasses must implement this method")
 
-    def execute(self, data: DataBundle, **kwargs: Any) -> DataBundle:
+    def execute(self, data: ProcessingData) -> None:
         """Execute the process step on the given data"""
+        self.processing_data = data
         if not self.__prepared:
             self.prepare_execution()
             self.__prepared = True
-        self.produced_outputs = self.calculate(data, **kwargs)
+        self.produced_outputs = self.calculate()
         for _key, value in self.produced_outputs.items():
-            data.data[_key] = value
+            if _key in data:
+                data[_key].update(value)
+            else:
+                data[_key] = value
         self.executed = True
-        return data
 
     def reset(self):
         """Reset the process step to its initial state"""
@@ -102,3 +148,36 @@ def modify_config(self, key: str, value: Any):
         else:
             raise KeyError(f"Key {key} not found in configuration")  # noqa
         self.__prepared = False
+
+    @classmethod
+    def is_process_step_dict(cls, instance: Type | None, attribute: str | None, item: Any) -> bool:
+        """
+        Check if the value is a dictionary with the correct keys and types.
+        """
+        if not isinstance(item, dict):
+            return False
+        for _key, _value in item.items():
+            if _key not in cls.CONFIG_KEYS:
+                return False
+            _config = cls.CONFIG_KEYS[_key]
+            if _value is None:
+                if _config["allow_none"]:
+                    continue
+                return False
+            if isinstance(_value, Iterable) and not isinstance(_value, str):
+                if not (
+                    _config["allow_iterable"]
+                    and all([isinstance(_i, _config["type"]) for _i in _value])
+                ):
+                    return False
+                continue
+            if not isinstance(_value, _config["type"]):
+                return False
+        return True
+
+    @classmethod
+    def default_config(cls) -> dict[str, Any]:
+        """
+        Create an initial dictionary for the process step configuration.
+        """
+        return {_k: _v["default"] for _k, _v in cls.CONFIG_KEYS.items()}

src/modacor/dataclasses/processing_data.py

@@ -0,0 +1,63 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright 2025 MoDaCor Authors
+#
+# Redistribution and use in source and binary forms, with or without modification,
+# are permitted provided that the following conditions are met:
+# 1. Redistributions of source code must retain the above copyright notice, this
+#    list of conditions and the following disclaimer.
+# 2. Redistributions in binary form must reproduce the above copyright notice,
+#    this list of conditions and the following disclaimer in the documentation
+#    and/or other materials provided with the distribution.
+# 3. Neither the name of the copyright holder nor the names of its contributors
+#    may be used to endorse or promote products derived from this software without
+#    specific prior written permission.
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS “AS IS” AND
+# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR
+# ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
+# ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+
+__all__ = ["ProcessingData"]
+__license__ = "BSD-3-Clause"
+__version__ = "0.0.1"
+
+from typing import Any
+
+from modacor.dataclasses.databundle import DataBundle
+
+
+class ProcessingData(dict):
+    """
+    The ProcessingData class is a dictionary-like object that stores reference
+    to DataBundles.
+    """
+
+    def __setitem__(self, key: str, item: DataBundle | Any):
+        """
+        Assign a value to a dictionary key.
+
+        Parameters
+        ----------
+        key : str
+            The dictionary key.
+        item : DataBundle | Any
+            The value / object to be added to the dictionary.
+
+        Raises
+        ------
+        TypeError
+            If the item is not an instance of DataBundle.
+        TypeError
+            If the key is not a string.
+        """
+        if not isinstance(item, DataBundle):
+            raise TypeError(f"Expected a DataBundle instance, got {type(item).__name__}.")
+        if not isinstance(key, str):
+            raise TypeError(f"Expected a string key, got {type(key).__name__}.")
+        super().__setitem__(key, item)