Merge pull request #49 from BAMresearch/static_data_iosource

Static data iosource, in the shape of YAMLLoader

Author: toqduj

docs/Classes.drawio

@@ -6,6 +6,8 @@
 from pint import UnitRegistry, set_application_registry
 
 ureg = UnitRegistry(system="SI")
+# we need to define an arbitrary intensity unit for scaling of intensity data:
+ureg.define("AIU = [intensity] = arbitrary_intensity_unit")
 Q_ = ureg.Quantity
 # recommended for pickling and unpickling:
 set_application_registry(ureg)

src/modacor/__init__.py

@@ -3,7 +3,7 @@
 
 import logging
 
-logger = logging.getLogger(__name__)
+# logger = logging.getLogger(__name__)
 
 
 class MessageHandler:
@@ -29,26 +29,26 @@ def __init__(self, level: int = logging.INFO, name: str = "MoDaCor", **kwargs):
         self.consoleLogHandler.setFormatter(formatter)
         self.logger.addHandler(self.consoleLogHandler)
 
-    def log(self, message: str, level: int = None, name: str = None):
+    def log(self, message: str, level: int = None) -> None:  # , name: str = None):
         if level is None:
             level = self.level
 
-        if name is None:
-            name = self.name
-
-        self.logger(message, level=level, name=name)
+        # if name is None:
+        #     name = self.name
+        # does not take a name: # self.logger = logging.getLogger(name)
+        self.logger.log(msg=message, level=level)
 
     def info(self, message: str):
-        self.log(message, level=logging.INFO, name="MoDaCor")
+        self.log(message, level=logging.INFO)
 
     def warning(self, message: str):
-        self.log(message, level=logging.WARNING, name="MoDaCor")
+        self.log(message, level=logging.WARNING)
 
     def error(self, message: str):
-        self.log(message, level=logging.ERROR, name="MoDaCor")
+        self.log(message, level=logging.ERROR)
 
     def critical(self, message: str):
-        self.log(message, level=logging.CRITICAL, name="MoDaCor")
+        self.log(message, level=logging.CRITICAL)
 
     def debug(self, message: str):
-        self.log(message, level=logging.DEBUG, name="MoDaCor")
+        self.log(message, level=logging.DEBUG)

src/modacor/dataclasses/messagehandler.py

@@ -1,38 +1,24 @@
 # 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.
-
-__license__ = "BSD-3-Clause"
-__copyright__ = "Copyright 2025 MoDaCor Authors"
-__status__ = "Alpha"
-__all__ = ["IoSource"]
-
-
-from typing import Any
+# /usr/bin/env python3
+# -*- coding: utf-8 -*-
 
+from __future__ import annotations
+
+__coding__ = "utf-8"
+__authors__ = ["Malte Storm", "Brian R. Pauw"]  # add names to the list as appropriate
+__copyright__ = "Copyright 2025, The MoDaCor team"
+__date__ = "14/06/2025"
+__status__ = "Development"  # "Development", "Production"
+# end of header and standard imports
+
+from typing import Any, Optional, Tuple, Union
+
+import numpy as np
 from attrs import define, field
 
-from modacor.dataclasses.source_data import SourceData
+# for type hinting of slicing:
+Index = Union[int, slice, type(Ellipsis)]
+ArraySlice = Union[Index, Tuple[Index, ...]]
 
 
 def default_config() -> dict[str, Any]:
@@ -78,21 +64,76 @@ class IoSource:
 
     configuration: dict[str, Any] = field(factory=default_config)
 
-    def get_data(self, index: int, data_key: str) -> SourceData:
+    def get_data(self, data_key: str, load_slice: Optional[ArraySlice] = None) -> np.ndarray:
         """
-        Get data and metadata from the IO source using the provided data key.
+        Get data from the IO source using the provided data key.
 
         Parameters
         ----------
-        index : int
-            The index to access the data.
         data_key : str
-            The key to access the data.
+            The key to access the data, e.g. '/entry1/instrument/detector00/data'.
+        load_slice : Optional[ArraySlice]
+            A slice or tuple of slices to apply to the data. If None, the entire data is returned.
+            Slicing is not yet implemented, so this will raise NotImplementedError if used.
+            Consider using the numpy.s_ or numpy.index_exp for simplifying the slicing syntax.
 
         Returns
         -------
         np.ndarray :
-            The data associated with the provided key.
+            The data array associated with the provided key. For scalars, this is a 0-d array.
+        """
+        if load_slice is not None:
+            raise NotImplementedError("Slicing is not yet implemented.")
+        raise NotImplementedError("This method should be implemented in subclasses.")
+
+    def get_data_shape(self, data_key: str) -> Tuple[int, ...]:
+        """
+        Get the shape of the data from the IO source if the format supports it else empty tuple.
+
+        Parameters
+        ----------
+        data_key : str
+            The key to the data for which the shape is requested.
+
+        Returns
+        -------
+        Tuple[int, ...] :
+            The shape of the data associated with the provided key.
+            Returns an empty tuple if nothing available or unsupported.
+        """
+        raise NotImplementedError("This method should be implemented in subclasses.")
+
+    def get_data_dtype(self, data_key: str) -> Optional[np.dtype]:
+        """
+        Get the data type of the data from the IO source if the format supports it else None.
+
+        Parameters
+        ----------
+        data_key : str
+            The key to the data for which the dtype is requested.
+
+        Returns
+        -------
+        Optional[np.dtype] :
+            The data type of the data associated with the provided key.
+            Returns None if nothing available or unsupported.
+        """
+        raise NotImplementedError("This method should be implemented in subclasses.")
+
+    def get_data_attributes(self, data_key: str) -> dict[str, Any]:
+        """
+        Get data attributes from the IO source if the format supports it else empty dict.
+
+        Parameters
+        ----------
+        data_key : str
+            The key to the data for which attributes are requested.
+
+        Returns
+        -------
+        dict[str, Any] :
+            The attributes associated with the data.
+            Returns an empty dictionary if nothing available or unsupported.
         """
         raise NotImplementedError("This method should be implemented in subclasses.")
 

src/modacor/dataclasses/source_data.py

@@ -132,6 +132,75 @@ def get_data(self, data_reference: str, index: int) -> np.ndarray:
         _source = self.get_source(_source_ref)
         return _source.get_data(index, _data_key)
 
+    def get_data_shape(self, data_reference: str, index: int) -> np.ndarray:
+        """
+        Get data from the specified source using the provided data key.
+
+        The data_reference is composed of the source reference and the internal
+        data reference, separated by "::".
+
+        Parameters
+        ----------
+        data_reference : str
+            The reference name of the source to access.
+        index : int
+            The index to access the data.
+
+        Returns
+        -------
+        Any :
+            The data associated with the provided key.
+        """
+        _source_ref, _data_key = self.split_data_reference(data_reference)
+        _source = self.get_source(_source_ref)
+        return _source.get_data_shape(index, _data_key)
+
+    def get_data_dtype(self, data_reference: str, index: int) -> np.ndarray:
+        """
+        Get data from the specified source using the provided data key.
+
+        The data_reference is composed of the source reference and the internal
+        data reference, separated by "::".
+
+        Parameters
+        ----------
+        data_reference : str
+            The reference name of the source to access.
+        index : int
+            The index to access the data.
+
+        Returns
+        -------
+        Any :
+            The data associated with the provided key.
+        """
+        _source_ref, _data_key = self.split_data_reference(data_reference)
+        _source = self.get_source(_source_ref)
+        return _source.get_data_dtype(index, _data_key)
+
+    def get_data_attributes(self, data_reference: str, index: int) -> np.ndarray:
+        """
+        Get data from the specified source using the provided data key.
+
+        The data_reference is composed of the source reference and the internal
+        data reference, separated by "::".
+
+        Parameters
+        ----------
+        data_reference : str
+            The reference name of the source to access.
+        index : int
+            The index to access the data.
+
+        Returns
+        -------
+        Any :
+            The data associated with the provided key.
+        """
+        _source_ref, _data_key = self.split_data_reference(data_reference)
+        _source = self.get_source(_source_ref)
+        return _source.get_data_attributes(index, _data_key)
+
     def get_static_metadata(self, data_reference: str) -> Any:
         """
         Get static metadata from the specified source using the provided data key.