Merge branch 'main' into pipeline_with_processsteps

Author: ahoermann

.flake8

@@ -1,5 +1,5 @@
 [flake8]
-ignore = E203, E266, E501, W503, F403, F401, F405
+ignore = E203, E266, E501, W503, F403, F401, F405, F821
 max-line-length = 120
 max-complexity = 18
 select = C,E,F,W,B,B950

LICENSE.txt

@@ -8,4 +8,4 @@ Redistribution and use in source and binary forms, with or without modification,
 
 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.
+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.

README.rst

@@ -66,8 +66,8 @@ https://BAMresearch.github.io/modacor
 Development
 ===========
 
-For coding contributions, we strongly recommend: 
-  - using flake8 and/or black for consistent formatting. 
+For coding contributions, we strongly recommend:
+  - using flake8 and/or black for consistent formatting.
   - writing tests for every added functionality -> towards test-driven coding practices.
 
 To run all the tests run::

ci/update.py

@@ -102,11 +102,7 @@ def main():
                     cov_report_path=project_meta["tool"]["coverage"]["report"]["path"],
                     # Python version to use for general tasks: docs (when tox did not set one)
                     py_ver=".".join(sys.version.split(".")[:2]),
-                    pypi_token=(
-                        "_".join(pypi_host + ["token"]).upper()
-                        if len(pypi_host)
-                        else "TEST_PYPI_TOKEN"
-                    ),
+                    pypi_token=("_".join(pypi_host + ["token"]).upper() if len(pypi_host) else "TEST_PYPI_TOKEN"),
                     pypi_repo="".join(pypi_host) if len(pypi_host) else "testpypi",
                 )
             )

docs/conf.py

@@ -36,9 +36,7 @@
 release = version
 commit_id = None
 try:
-    commit_id = (
-        subprocess.check_output(["git", "rev-parse", "--short", "HEAD"]).strip().decode("ascii")
-    )
+    commit_id = subprocess.check_output(["git", "rev-parse", "--short", "HEAD"]).strip().decode("ascii")
 except subprocess.CalledProcessError as e:
     print(e)
 

pyproject.toml

@@ -99,12 +99,12 @@ exclude_commit_patterns = ["chore", ".*\\bGHA\\b.*", ".*\\b[gG][hH] actions?\\b.
 upload_to_vcs_release = false
 
 [tool.black]
-line-length = 100
+line-length = 120
 preview = true
 
 [tool.isort]
 profile = "black"
-line_length = 100
+line_length = 120
 group_by_package = true
 known_first_party = "modacor"
 ensure_newline_before_comments = true

src/modacor/dataclasses/basedata.py

@@ -1,13 +1,16 @@
+__all__ = ["BaseData"]
+
 # import tiled
 # import tiled.client
 import logging
 from typing import Dict, List, Optional, Self
 
 import numpy as np
-import pint
 from attrs import define, field
 from attrs import validators as v
 
+from modacor import ureg
+
 logger = logging.getLogger(__name__)
 
 
@@ -20,8 +23,7 @@ def validate_rank_of_data(instance, attribute, value):
     # This assumes that signal is provided and is a valid numpy array.
     if instance.signal is not None and value > instance.signal.ndim:
         raise ValueError(
-            f"{attribute.name} ({value}) cannot exceed the dimensionality of signal "
-            f"(ndim={instance.signal.ndim})."
+            f"{attribute.name} ({value}) cannot exceed the dimensionality of signal (ndim={instance.signal.ndim})."
         )
 
 
@@ -32,13 +34,8 @@ class BaseData:
     It is designed to be used as a base class for more specialized data classes.
     """
 
-    # Unit information using Pint units - required input (ingest, internal, and display)
-    ingest_units: pint.Unit = field(validator=v.instance_of(pint.Unit))
-    internal_units: pint.Unit = field(validator=v.instance_of(pint.Unit))
-    display_units: pint.Unit = field(validator=v.instance_of(pint.Unit))
-
     # 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)])
@@ -50,24 +47,30 @@ class BaseData:
     rank_of_data: int = field(factory=int, validator=[v.instance_of(int), validate_rank_of_data])
 
     # Scalers to put on the denominator, sparated from the array for distinct uncertainty
-    normalization: Optional[np.ndarray] = field(
-        default=None, validator=v.optional(v.instance_of(np.ndarray))
-    )
+    normalization: Optional[np.ndarray] = field(default=None, validator=v.optional(v.instance_of(np.ndarray)))
     normalization_factor: float = field(default=1.0, validator=v.instance_of(float))
     normalization_factor_variance: float = field(default=0.0, validator=v.instance_of(float))
-    normalization_units: pint.Unit = field(
-        default=pint.Unit("dimensionless"), validator=v.instance_of(pint.Unit)
+    # Unit information using Pint units - required input (ingest, internal, and display)
+    signal_units: ureg.Unit = field(
+        default=ureg.Unit("dimensionless"), validator=v.instance_of(ureg.Unit)
+    )
+
+    normalization_units: ureg.Unit = field(
+        default=ureg.Unit("dimensionless"), validator=v.instance_of(ureg.Unit)
     )
-    normalization_factor_units: pint.Unit = field(
-        default=pint.Unit("dimensionless"), validator=v.instance_of(pint.Unit)
+    normalization_factor_units: ureg.Unit = field(
+        default=ureg.Unit("dimensionless"), validator=v.instance_of(ureg.Unit)
     )
     # array with some normalization (exposure time, solid-angle ....)
 
+    @property
+    def shape(self):
+        return self.signal.shape
+
     def __attrs_post_init__(self):
         if self.normalization is None:
-            self.normalization = np.ones(self.signal.shape)
+            self.normalization = np.ones(self.shape)
 
-    @property
     def mean(self) -> np.ndarray:
         """
         Returns the signal array with the normalization applied.
@@ -82,13 +85,6 @@ def std(self, kind) -> np.ndarray:
         """
         return np.sqrt(self.variances[kind] / self.normalization)
 
-    def sem(self, kind) -> np.ndarray:
-        """
-        Returns the uncertainties, i.e. standard deviation
-        The result is cast to internal units.
-        """
-        return np.sqrt(self.variances[kind]) / self.normalization
-
     @property
     def _unit_scale(self, display_units) -> float:
         return (1 * self.internal_units).to(display_units).magnitude

src/modacor/dataclasses/integrated_data.py

@@ -24,8 +24,6 @@ class IntegratedData(BaseData):
     \left( \frac{\sigma_A^2}{\mu_A^2} + \frac{\sigma_B^2}{\mu_B^2} \right)
     ```
     """
-
-    average: np.ndarray = field(factory=np.ndarray, validator=[v.instance_of(np.ndarray)])
     std: Dict[str, np.ndarray] = field(factory=dict, validator=[v.instance_of(dict)])
     sem: Dict[str, np.ndarray] = field(factory=dict, validator=[v.instance_of(dict)])
     # Core data array stored as an xarray DataArray
@@ -36,6 +34,4 @@ class IntegratedData(BaseData):
 
     # array with some normalization (exposure time, solid-angle ....)
     sum_normalization: np.ndarray = field(factory=np.ndarray, validator=[v.instance_of(np.ndarray)])
-    sum_normalization_squared: np.ndarray = field(
-        factory=np.ndarray, validator=[v.instance_of(np.ndarray)]
-    )
+    sum_normalization_squared: np.ndarray = field(factory=np.ndarray, validator=[v.instance_of(np.ndarray)])

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)
@@ -48,12 +91,19 @@ class ProcessStep:
 
     # a message handler, supporting logging, warnings, errors, etc. emitted by the process
     # during execution
-    message_handler: MessageHandler = field(
-        default=MessageHandler(), validator=v.instance_of(MessageHandler)
-    )
+    message_handler: MessageHandler = field(default=MessageHandler(), validator=v.instance_of(MessageHandler))
 
     # 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()
 
     # add hash function. equality can be checked
     def __hash__(self):
@@ -68,30 +118,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[_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"""
@@ -106,3 +150,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/process_step_describer.py

@@ -33,17 +33,13 @@ def validate_required_data_keys(instance, attribute, value):
 @define
 class ProcessStepDescriber:
     calling_name: str = field()  # short name to identify the calling process for the UI
-    calling_id: str = (
-        field()
-    )  # not sure what we were planning here. some UID perhaps? difference with calling_module
+    calling_id: str = field()  # not sure what we were planning here. some UID perhaps? difference with calling_module
     calling_module_path: Path = field(
         validator=v.instance_of(Path)
     )  # partial path to the module from src/modacor/modules onwards
     calling_version: str = field()  # module version being executed
     required_data_keys: list[str] = field(factory=list)  # list of data keys required by the process
-    required_arguments: list[str] = field(
-        factory=list
-    )  # list of argument key-val combos required by the process
+    required_arguments: list[str] = field(factory=list)  # list of argument key-val combos required by the process
     calling_arguments: dict[str, Any] = field(factory=dict, validator=validate_required_keys)
     works_on: dict[str, list] = field(
         factory=dict, validator=v.instance_of(dict)