update defaults behaviour

src/hdfmap/eval_functions.py

@@ -13,13 +13,15 @@
 import h5py
 from types import EllipsisType
 
+from .objects import Dataset
 from .logging import create_logger
 
 # parameters
 GLOBALS_NAMELIST = asteval.make_symbol_table(use_numpy=True).keys()
 DEFAULT: typing.Any = np.array('--')  # default return in eval
 SEP = '/'  # HDF path separator
 OMIT = ['/value', '/data']  # omit these names in paths when determining identifier
+LOCAL_NAME = 'local_name'  # dataset attribute name for alt_name (DLS specific)
 logger = create_logger(__name__)
 # regex patterns
 re_special_characters = re.compile(r'\W')  # finds all special non-alphanumberic characters
@@ -33,6 +35,16 @@
     logger.warning("Nexus timestamps are not convertable by datetime.fromisoformat in python version <3.11")
 
 
+def generate_alt_name(hdf_dataset: h5py.Dataset) -> str | None:
+    """Generate alt_name of dataset if 'local_name' in attributes"""
+    if LOCAL_NAME in hdf_dataset.attrs:
+        alt_name = hdf_dataset.attrs[LOCAL_NAME]
+        if hasattr(alt_name, 'decode'):
+            alt_name = alt_name.decode()
+        return expression_safe_name(alt_name.split('.')[-1])
+    return None
+
+
 def generate_identifier(hdf_path: str | bytes) -> str:
     """
     Generate a valid python identifier from a hdf dataset path or other string
@@ -223,7 +235,7 @@ def dataset2str(dataset: h5py.Dataset, index: int | slice | tuple[slice, ...] =
             return str(np.squeeze(dataset[index]))  # other np.ndarray
 
 
-def dataset_attribute(dataset: h5py.Dataset, attribute: str) -> str:
+def dataset_attribute(dataset: h5py.Dataset | Dataset, attribute: str) -> str:
     """
     Return attribute of dataset
     """
@@ -266,6 +278,36 @@ def find_or_expressions(expr: str) -> set[str]:
     return set(alternates)
 
 
+def replace_or_expressions(expr: str, path_map: dict[str, str], path_check: h5py.File | list | dict,
+                           data: dict[str, typing.Any]) -> str:
+    """
+    Replaces instances of (a|b|c)-OR-expressions within the string.
+    The first item of the OR-expression that is found in data or if the
+    associated path is found in path_check will replace the expression.
+
+        path_map = {'x': '/entry/data/x', 'y': '/entry/data/y'}
+        path_check = h5py.File('mydata.h5')  # file contains x path but not y path
+        expression = '(y|x)'
+        new = replace_or_expressions(expression, path_map, path_check, {})
+        # new == 'x'
+
+    :param expr: string expression with components to replace
+    :param path_map: dict mapping from variable name to path as found in path_check
+    :param path_check: Hdf file to check path in, or could be a list or dict of paths.
+    :param data: dict mapping of variable name to data
+    :return: expression with replaced components
+    """
+    while or_expressions := find_or_expressions(expr):
+        for or_exp in or_expressions:
+            names = or_exp.strip('()').split('|')
+            name = next(
+                (n for n in names if n in data),
+                next((n for n in names if path_map.get(n, '') in path_check), names[-1])
+            )
+            expr = expr.replace(or_exp, name)
+    return expr
+
+
 def replace_expression_vars(expr: str, mapping: dict[str, str]) -> str:
     """
     Replace variable names in an expression
@@ -293,6 +335,57 @@ def replace_expression_vars(expr: str, mapping: dict[str, str]) -> str:
     return expr
 
 
+def replace_defaults(expr: str, path_map: dict[str, str]) -> str:
+    """
+    Replace default specifiers in an expression
+    A default specifier is given as 'a?(b)', where a will be
+    returned if 'a' is in path_map, otherwise 'b' is returned.
+
+        path_map = {'x': '/entry/data/x', 'y': '/entry/data/y'}
+        expression = 'x + y?(10) + z?(20)'
+        result = replace_defaults(expression, path_map)
+        # result == 'x + y + 20'
+
+    :param expr: string expression with components to replace
+    :param path_map: dict mapping from variable name to path
+    :return: expression with replaced components
+    """
+    for match in re_dataset_default.finditer(expr):
+        name, name_default = match.groups()
+        if name not in path_map:
+            expr = expr.replace(match.group(), name_default)
+        else:
+            expr = expr.replace(match.group(), name)
+    return expr
+
+
+def replace_attributes(expr: str, path_map: dict[str, str],
+                       hdf_file: h5py.File | dict[str, Dataset]) -> tuple[str, dict]:
+    """
+    Find sub-strings of the form 'name@attribute', and replace this with
+    the attribute associated with the dataset specified by name.
+
+        path_map = {'x': '/entry/data/x', 'y': '/entry/data/y'}
+        hdf_file = h5py.File('mydata.h5'), where dataset('/entry/data/x') as attribute 'units'='mm'
+        expression = 'x x@units'
+        result = replace_attributes(expression, path_map, hdf_file)
+        # result == 'x mm'
+
+    :param expr: string expression with components to replace
+    :param path_map: dict mapping from variable name to path
+    :param hdf_file: h5py.File object
+    :return: expression with replaced components
+    """
+    attributes = {
+        f"attr__{name}_{attr}": dataset_attribute(hdf_file[path], attr)
+        for name, attr in re_dataset_attributes.findall(expr)  # name@attr
+        if (path := path_map.get(name, '')) in hdf_file
+    }
+    # replace name@attribute in expression
+    expr = re_dataset_attributes.sub(r'attr__\g<1>_\g<2>', expr)
+    return expr, attributes
+
+
 def extra_hdf_data(hdf_file: h5py.File) -> dict:
     """Extract filename, filepath and other additional data fom hdf file"""
     filepath = getattr(hdf_file, 'filename', 'unknown')
@@ -385,36 +478,17 @@ def prepare_expression(hdf_file: h5py.File, expression: str, hdf_namespace: dict
     :param replace_names: dict of {'variable_name': expression}
     :return: str expression
     """
+    if data_namespace is None:
+        data_namespace = {}  # note the link to the parent namespace must be preserved
     # replace names with expressions
     expression = replace_expression_vars(expression, replace_names)
-
-    if data_namespace is None:
-        data_namespace = {}
-    # find name@attribute in expression
-    attributes = {
-        f"attr__{name}_{attr}": dataset_attribute(hdf_file[path], attr)
-        for name, attr in re_dataset_attributes.findall(expression)  # name@attr
-        if (path := hdf_namespace.get(name, '')) in hdf_file
-    }
+    # replace name@attribute in expression and add values to data_namespace
+    expression, attributes = replace_attributes(expression, hdf_namespace, hdf_file)
     data_namespace.update(attributes)  # adds data in the parent function
-    # replace name@attribute in expression
-    expression = re_dataset_attributes.sub(r'attr__\g<1>_\g<2>', expression)
     # find values with defaults '..?(..)'
-    for match in re_dataset_default.finditer(expression):
-        name, name_default = match.groups()
-        if name not in hdf_namespace:
-            expression = expression.replace(match.group(), name_default)
-        else:
-            expression = expression.replace(match.group(), name)
+    expression = replace_defaults(expression, hdf_namespace)
     # find alternate names '(opt1|opt2|opt3)'
-    while or_expressions := find_or_expressions(expression):
-        for or_exp in or_expressions:
-            names = or_exp.strip('()').split('|')
-            name = next(
-                (n for n in names if n in attributes),
-                next((n for n in names if hdf_namespace.get(n, '') in hdf_file), names[-1])
-            )
-            expression = expression.replace(or_exp, name)
+    expression = replace_or_expressions(expression, hdf_namespace, hdf_file, attributes)
     return expression
 
 
@@ -544,37 +618,3 @@ def format_hdf(hdf_file: h5py.File, expression: str, hdf_namespace: dict[str, st
     )
 
 
-class HdfMapInterpreter(asteval.Interpreter):
-    """
-    HdfMap implementation of asteval.Interpreter
-
-    Expression is parsed for patterns and loads HDF data before evaluation
-
-        m = HdfMap('file.nxs')
-        ii = HdfMapInterpreter(m, replace_names={}, default='', **kwargs)
-        out = ii.eval('expression')
-
-    :param hdfmap: HdfMap instance (including hdfmap.filename pointing to the HDF file)
-    :param replace_names: dict of {'variable_name': expression}
-    :param default: returned if varname not in namespace
-    :param kwargs: keyword arguments passed to asteval.Interpreter
-    """
-    def __init__(self, hdfmap, replace_names: dict[str, str], default: typing.Any = DEFAULT, **kws):
-        super().__init__(**kws)
-        self.hdfmap = hdfmap
-        self.replace_names = replace_names
-        self.default_value = default
-        self.use_stored_data = False
-
-    def eval(self, expr, lineno=0, show_errors=True, raise_errors=False):
-        with self.hdfmap.load_hdf() as hdf:
-            new_expression = prepare_expression_load_data(
-                hdf_file=hdf,
-                expression=expr,
-                hdf_namespace=self.hdfmap.combined,
-                data_namespace=self.symtable,
-                replace_names=self.replace_names,
-                default=self.default_value,
-                use_stored_data=self.use_stored_data
-            )
-        return super().eval(new_expression, lineno, show_errors, raise_errors)

src/hdfmap/hdfmap_class.py

@@ -3,6 +3,7 @@
 """
 import json
 import typing
+import re
 from collections import defaultdict
 
 import numpy as np
@@ -11,49 +12,20 @@
 from . import load_hdf
 from .data_holder import disp_dict, DataHolder
 from .logging import create_logger
-from .eval_functions import (expression_safe_name, extra_hdf_data, eval_hdf, HdfMapInterpreter,
+from .eval_functions import (extra_hdf_data, eval_hdf, prepare_expression_load_data,
                              format_hdf, dataset2data, dataset2str, is_image, attrs2dict,
-                             DEFAULT, SEP, generate_identifier, build_hdf_path)
-
+                             DEFAULT, SEP, generate_identifier, build_hdf_path, generate_alt_name,
+                             replace_or_expressions, replace_expression_vars, replace_defaults,
+                             find_identifiers)
+from .objects import Group, Dataset
 
 # parameters
-LOCAL_NAME = 'local_name'  # dataset attribute name for alt_name
 IMAGE_DATA = 'IMAGE'  # namespace name for default image data
 
 # logger
 logger = create_logger(__name__)
 
 
-class Group(typing.NamedTuple):
-    nx_class: str
-    name: str
-    attrs: dict
-    datasets: list[str]
-    parent: "Group | None"
-    default: bool
-    external_file: str | None
-
-
-class Dataset(typing.NamedTuple):
-    name: str
-    names: list[str]
-    size: int
-    shape: tuple[int]
-    attrs: dict
-    parent: Group
-    external_file: str | None
-
-
-def generate_alt_name(hdf_dataset: h5py.Dataset) -> str | None:
-    """Generate alt_name of dataset if 'local_name' in attributes"""
-    if LOCAL_NAME in hdf_dataset.attrs:
-        alt_name = hdf_dataset.attrs[LOCAL_NAME]
-        if hasattr(alt_name, 'decode'):
-            alt_name = alt_name.decode()
-        return expression_safe_name(alt_name.split('.')[-1])
-    return None
-
-
 class HdfMap:
     """
     HdfMap object, container for paths of different objects in an HDF file
@@ -621,7 +593,30 @@ def first_last_scannables(self, last_n: int = 1) -> tuple[dict[str, str], dict[s
         last = {name: self.scannables[name] for name in scannable_names[::-1][:last_n]}
         return first, last
 
-    def get_path(self, name_or_path):
+    def get_default_name(self, name_or_path: str) -> str | None:
+        """Return the default name of a dataset or group identifier"""
+        path = self.get_path(name_or_path)
+        if path and path in self.datasets:
+            return self.datasets[path].name
+        elif path and path in self.groups:
+            return self.groups[path].name
+        return None
+
+    def merge_default_names(self, expression: str) -> str:
+        """Replace names in the expression with dataset default names"""
+        expression = replace_expression_vars(expression, self.alternate_names)
+        if expression.startswith('/'):
+            return self.get_default_name(expression) or expression
+        expression = replace_defaults(expression, self.combined)
+        expression = replace_or_expressions(expression, self.combined, self.datasets, {})
+        ids = find_identifiers(expression)
+        if not ids:
+            return expression
+        default_ids = {name: self.get_default_name(name) or name for name in ids}
+        re_ids = re.compile("|".join(default_ids))
+        return re_ids.sub(lambda m: default_ids[m.group(0)], expression)
+
+    def get_path(self, name_or_path) -> str | None:
         """Return hdf path of object in HdfMap"""
         if name_or_path in self.datasets or name_or_path in self.groups:
             return name_or_path
@@ -837,9 +832,14 @@ def generate_ids(self, *names: str, modify_missing: bool = True) -> list[str]:
         Will return the path identifier of the given name if the name is in the namespace,
         otherwise a valid identifier will be generated.
 
-            xlabel, ylabel = generate_axis_labels('axes', 'signal')
-            generate_axis_labels('my/data/label', modify_missing=True) #-> ['label', ]
-            generate_axis_labels('(x-y)/y', modify_missing=False) #-> ['(x-y)/y', ]
+            xlabel, ylabel = generate_ids('axes', 'signal')
+            generate_ids('my/data/label', modify_missing=True) #-> ['label', ]
+            generate_ids('(x-y)/y', modify_missing=False) #-> ['(x-y)/y', ]
+
+        See also: get_default_name
+
+        Both generate_ids and get_default_name return names from paths or alternate names,
+        however generate_ids just takes the last element of the path.
 
         :param names: names to generate axis labels for
         :param modify_missing: if True, modifies names even if they are not in namespace
@@ -1095,7 +1095,7 @@ def get_dataholder(self, hdf_file: h5py.File, flatten_scannables: bool = False)
 
     def eval(self, hdf_file: h5py.File, expression: str, default=DEFAULT,
              local_data: dict | None = None, prefer_local: bool | None = None,
-             raise_errors: bool = True):
+             raise_errors: bool = True) -> typing.Any:
         """
         Evaluate an expression using the namespace of the hdf file
         :param hdf_file: h5py.File object
@@ -1141,30 +1141,37 @@ def format_hdf(self, hdf_file: h5py.File, expression: str, default=DEFAULT,
             raise_errors=raise_errors
         )
 
-    def create_interpreter(self, default=DEFAULT, local_data: dict | None = None, prefer_local: bool | None = None):
+    def generate_eval_expression(self, hdf_file: h5py.File, expression: str, default=DEFAULT,
+                                 local_data: dict | None = None, prefer_local: bool | None = None) -> tuple[str, dict[str, typing.Any]]:
         """
-        Create an interpreter object for the current file
-        The interpreter is a sub-class of asteval.Interpreter that parses expressions for hdfmap eval patters
-        and loads data when required.
+        Evaluate an expression using the namespace of the hdf file,
+        returning the evaluated expression and dictionary of data for identifiers
 
-        The hdf file self.filename is used to extract data and is only opened during evaluation.
+            expression, data = HdfMap.generate_eval_expression(hdf, 'signal / (monitor|ic1monitor)')
 
-            ii = HdfMap.create_interpreter()
-            out = ii.eval('expression')
+        This function serves as a useful way to debug expressions for the eval_hdf function.
+        Note that the hdf file object must be included as the way the expression is evaluated
+        means individual expression components are checked against the HdfMap namespace and
+        the hdf file, allowing lazy loading of data (loading only the data needed).
 
+        :param hdf_file: h5py.File object
+        :param expression: str expression to be evaluated
         :param default: returned if varname not in namespace
-        :param local_data: replaces the HdfMap local_data attribute for this file
+        :param local_data: dict of additional data to pass to the expression, as {'varname': data}
         :param prefer_local: uses values in local_data first if available when True
+        :return: expression, dict - data namespace
         """
-        interpreter = HdfMapInterpreter(
-            hdfmap=self,
+        local_data = self._local_data.copy() if local_data is None else local_data.copy()
+        new_expression = prepare_expression_load_data(
+            hdf_file=hdf_file,
+            expression=expression,
+            hdf_namespace=self.combined,
+            data_namespace=local_data,
             replace_names=self.alternate_names,
             default=default,
-            user_symbols=self._local_data if local_data is None else local_data,
-            use_numpy=True
+            use_stored_data=self._use_local_data if prefer_local is None else prefer_local,
         )
-        interpreter.use_stored_data = self._use_local_data if prefer_local is None else prefer_local
-        return interpreter
+        return new_expression, local_data
 
     def create_dataset_summary(self, hdf_file: h5py.File) -> str:
         """Create summary of all datasets in file"""