Config dict

ifsbench/__init__.py

@@ -16,6 +16,7 @@
 from .arch import * # noqa
 from .benchmark import * # noqa
 from .cli import * # noqa
+from .config_mixin import * # noqa
 from .darshanreport import * # noqa
 from .drhook import * # noqa
 from .files import * # noqa

ifsbench/config_mixin.py

@@ -0,0 +1,117 @@
+from abc import ABC, abstractmethod
+from typing import Any, Dict, get_args, get_origin, get_type_hints, List, Union
+
+__all__ = ['ConfigMixin', 'CONF']
+
+
+CONF = Union[int, float, str, bool, Dict, List, None]
+
+
+def _config_from_locals(config: Dict[str, Any]) -> None:
+    config = config.copy()
+    config.pop('self', None)
+    config.pop('cls', None)
+    return config
+
+
+class ConfigMixin(ABC):
+    """
+    Base class for handling configurations in a format that can be used for storage.
+
+    The contents of the config are based on the parameters required by the implementing
+    classes constructor. Because of this, additional entries cannot be added to an existing config.
+    However, the values of individual entries can be updated with a value of the same type.
+
+    The required format can be either created based on the constructor, or explicitly set by
+    implementing the `config_format` method.
+
+    Parameters
+    ----
+    config: dictionary containing parameter names and their values
+    """
+
+    _mixin_config = None
+
+    @classmethod
+    @abstractmethod
+    def config_format(cls) -> Dict[str, Any]:
+        raise NotImplementedError()
+
+    @classmethod
+    def _format_from_init(cls) -> Dict[str, Any]:
+        format_definition = dict(get_type_hints(cls.__init__, include_extras=False))
+        format_definition = _config_from_locals(format_definition)
+        return format_definition
+
+    def set_config_from_init_locals(self, config: Dict[str, Any]):
+        config = _config_from_locals(config)
+        self.set_config(config)
+
+    def set_config(self, config: Dict[str, CONF]) -> None:
+        if self._mixin_config:
+            raise ValueError('Config already set.')
+        self._mixin_config = config
+
+    def get_config(self) -> Dict[str, CONF]:
+        return self._mixin_config
+
+    def update_config(self, field: str, value: CONF) -> None:
+        if field not in self._mixin_config:
+            raise ValueError(f'{field} not part of config {self._mixin_config}, not setting')
+        if not isinstance(value, type(self._mixin_config[field])):
+            raise ValueError(
+                f'Cannot update config: wrong type {type(value)} for field {field}'
+            )
+        self._mixin_config[field] = value
+
+    @classmethod
+    def validate_config(cls, config: Dict[str, CONF]):
+        format_definition = cls.config_format()
+        cls._validate_config_from_format(config, format_definition)
+
+    @classmethod
+    def _validate_config_from_format(
+        cls, config: Dict[str, CONF], format_definition: Dict[str, Any]
+    ):
+
+        for key, value in config.items():
+            if not isinstance(value, CONF):
+                # check that the given value is a valid config type
+                raise ValueError(f'Unsupported config value type for {value}')
+            if key not in format_definition:
+                raise ValueError(f'unexpected key "{key}" in config, expected {format_definition}')
+
+        for key, value in format_definition.items():
+
+            if (key not in config) and (type(None) not in get_args(value)):
+                # format_definition key has to be in config unless it's optional
+                raise ValueError(f'"{key}" required but not in {config}')
+            if isinstance(value, Dict):
+                # nested, check that field also nested in config, then recursively check dict.
+                if not isinstance(config[key], Dict):
+                    raise ValueError(
+                        f'"{key}" has type {type(config[key])}, expected {value}'
+                    )
+                cls._validate_config_from_format(config[key], format_definition[key])
+            elif isinstance(value, List):
+                # For now, only check both are lists and first entry type is correct, don't check every entry.
+                if not isinstance(config[key], list):
+                    raise ValueError(
+                        f'"{key}" has type {type(config[key])}, expected {value}'
+                    )
+                if not isinstance(value[0], type(config[key][0])):
+                    raise ValueError(
+                        f'list entries for "{key}" have type {type(config[key][0])}, expected {type(value[0])}'
+                    )
+            elif get_origin(value) == Union and type(None) in get_args(value):
+                # Optional: check matching type or None
+                opt_type = get_args(value)
+                if key in config and type(config[key]) not in opt_type:
+                    raise ValueError(
+                        f'wrong type for optional {type(value)}: {config[key]}'
+                    )
+            elif not isinstance(config[key], value):
+                # types of format and config have to match
+                raise ValueError(
+                    f'"{key}" has type {type(config[key])}, expected {value}'
+                )

ifsbench/data/extracthandler.py

@@ -7,14 +7,16 @@
 
 import pathlib
 import shutil
+from typing import Any, Dict, Optional
 
+from ifsbench.config_mixin import CONF, ConfigMixin
 from ifsbench.data.datahandler import DataHandler
 from ifsbench.logging import debug
 
 __all__ = ['ExtractHandler']
 
 
-class ExtractHandler(DataHandler):
+class ExtractHandler(DataHandler, ConfigMixin):
     """
     DataHandler that extracts a given archive to a specific directory.
 
@@ -31,13 +33,25 @@ class ExtractHandler(DataHandler):
         :meth:`execute`.
     """
 
-    def __init__(self, archive_path, target_dir=None):
+    def __init__(self, archive_path: str, target_dir: Optional[str] = None):
+        self.set_config_from_init_locals(locals())
         self._archive_path = pathlib.Path(archive_path)
         if target_dir is None:
             self._target_dir = None
         else:
             self._target_dir = pathlib.Path(target_dir)
 
+    @classmethod
+    def config_format(cls) -> Dict[str, Any]:
+        return cls._format_from_init()
+
+    @classmethod
+    def from_config(cls, config: dict[str, CONF]) -> 'ExtractHandler':
+        cls.validate_config(config)
+        archive_path = config['archive_path']
+        target_dir = config['target_dir'] if 'target_dir' in config else None
+        return cls(archive_path, target_dir)
+
     def execute(self, wdir, **kwargs):
         wdir = pathlib.Path(wdir)
 
@@ -46,7 +60,7 @@ def execute(self, wdir, **kwargs):
             if self._target_dir.is_absolute():
                 target_dir = self._target_dir
             else:
-                target_dir = wdir/self._target_dir
+                target_dir = wdir / self._target_dir
 
         debug(f"Unpack archive {self._archive_path} to {target_dir}.")
         shutil.unpack_archive(self._archive_path, target_dir)

ifsbench/data/namelisthandler.py

@@ -5,23 +5,28 @@
 # granted to it by virtue of its status as an intergovernmental organisation
 # nor does it submit to any jurisdiction.
 
-from enum import auto, Enum
+from enum import auto, StrEnum
 import pathlib
+from typing import Any, Dict
+
 
 import f90nml
 
+from ifsbench.config_mixin import CONF, ConfigMixin
 from ifsbench.data.datahandler import DataHandler
 from ifsbench.logging import debug, info
 
 
 __all__ = ['NamelistOverride', 'NamelistHandler', 'NamelistOperation']
 
-class NamelistOperation(Enum):
+
+class NamelistOperation(StrEnum):
     SET = auto()
     APPEND = auto()
     DELETE = auto()
 
-class NamelistOverride:
+
+class NamelistOverride(ConfigMixin):
     """
     Specify changes that will be applied to a namelist.
 
@@ -42,23 +47,48 @@ class NamelistOverride:
         The value that is set (SET operation) or appended (APPEND).
     """
 
+    def __init__(
+        self, namelist: str, entry: str, mode: NamelistOperation, value: CONF = None
+    ):
 
-    def __init__(self, key, mode, value=None):
-        if isinstance(key, str):
-            self._keys = key.split('/')
-        else:
-            self._keys = tuple(key)
-
-        if len(self._keys) != 2:
-            raise ValueError("The key object must be of length two.")
-
+        self.set_config_from_init_locals(locals())
+        self._keys = (namelist, entry)
         self._mode = mode
         self._value = value
 
         if self._value is None:
             if self._mode in (NamelistOperation.SET, NamelistOperation.APPEND):
                 raise ValueError("The new value must not be None!")
 
+    @classmethod
+    def from_keytuple(
+        cls, key: tuple[str, str], mode: NamelistOperation, value: CONF = None
+    ) -> 'NamelistOverride':
+        if len(key) != 2:
+            raise ValueError(f"The key tuple must be of length two, found key {key}.")
+        return cls(key[0], key[1], mode, value)
+
+    @classmethod
+    def from_keystring(
+        cls, key: str, mode: NamelistOperation, value: CONF = None
+    ) -> 'NamelistOverride':
+        keys = key.split('/')
+        if len(keys) != 2:
+            raise ValueError(
+                f"The key string must contain single '/', found key {key}."
+            )
+        return cls(keys[0], keys[1], mode, value)
+
+    @classmethod
+    def from_config(cls, config: dict[str, CONF]) -> 'NamelistOverride':
+        cls.validate_config(config)
+        value = config['value'] if 'value' in config else None
+        return cls(config['namelist'], config['entry'], config['mode'], value)
+
+    @classmethod
+    def config_format(cls) -> Dict[str, Any]:
+        return cls._format_from_init()
+
     def apply(self, namelist):
         """
         Apply the stored changes to a namelist.
@@ -98,7 +128,9 @@ def apply(self, namelist):
                 type_value = type(self._value)
 
                 if type_list != type_value:
-                    raise ValueError("The given value must have the same type as existing array entries!")
+                    raise ValueError(
+                        "The given value must have the same type as existing array entries!"
+                    )
 
             debug(f"Append {str(self._value)} to namelist entry {str(self._keys)}.")
 
@@ -109,7 +141,8 @@ def apply(self, namelist):
                 debug(f"Delete namelist entry {str(self._keys)}.")
                 del namelist[key]
 
-class NamelistHandler(DataHandler):
+
+class NamelistHandler(DataHandler, ConfigMixin):
     """
     DataHandler specialisation that can modify Fortran namelists.
 
@@ -129,7 +162,18 @@ class NamelistHandler(DataHandler):
         The NamelistOverrides that will be applied.
     """
 
-    def __init__(self, input_path, output_path, overrides):
+    def __init__(
+        self, input_path: str, output_path: str, overrides: list[NamelistOverride]
+    ):
+
+        override_confs = [no.get_config() for no in overrides]
+        self.set_config(
+            {
+                'input_path': input_path,
+                'output_path': output_path,
+                'overrides': override_confs,
+            }
+        )
 
         self._input_path = pathlib.Path(input_path)
         self._output_path = pathlib.Path(output_path)
@@ -139,13 +183,34 @@ def __init__(self, input_path, output_path, overrides):
             if not isinstance(override, NamelistOverride):
                 raise ValueError("Namelist overrides must be NamelistOverride objects!")
 
+    @classmethod
+    def config_format(cls) -> dict[str, Any]:
+        return {
+            'input_path': str,
+            'output_path': str,
+            'overrides': [
+                {
+                    str: CONF,
+                },
+            ],
+        }
+
+    @classmethod
+    def from_config(cls, config: dict[str, CONF]) -> 'NamelistHandler':
+        cls.validate_config(config)
+        input_path = config['input_path']
+        output_path = config['output_path']
+        override_configs = config['overrides']
+        overrides = [NamelistOverride.from_config(oc) for oc in override_configs]
+        return cls(input_path, output_path, overrides)
+
     def execute(self, wdir, **kwargs):
         wdir = pathlib.Path(wdir)
 
         if self._input_path.is_absolute():
             input_path = self._input_path
         else:
-            input_path = wdir/self._input_path
+            input_path = wdir / self._input_path
 
         # Do nothing if the input namelist doesn't exist.
         if not input_path.exists():
@@ -155,7 +220,7 @@ def execute(self, wdir, **kwargs):
         if self._output_path.is_absolute():
             output_path = self._output_path
         else:
-            output_path = wdir/self._output_path
+            output_path = wdir / self._output_path
 
         debug(f"Modify namelist {input_path}.")
         namelist = f90nml.read(input_path)